spb/uqo-research-db
Public
1# ============================================================2# Project: EDGAR Research Database3# Description: SEC EDGAR Research Infrastructure4#5# Authors:6# - Manel Kammoun — manel.kammoun@uqo.ca7# - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca8# - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca9#10# Institution:11# Universite du Quebec en Outaouais (UQO)12#13# License:14# Research / Academic Use15# ============================================================1617from __future__ import annotations1819import json20from pathlib import Path21from typing import TYPE_CHECKING2223import numpy as np24import pandas as pd2526from src.utils.constants import PROJECT_ROOT27from src.utils.logging import get_logger2829if TYPE_CHECKING:30 from src.storage.database import DatabaseManager3132logger = get_logger("factors.diagnostics")333435class FactorDiagnostics:3637 def __init__(self, db: DatabaseManager) -> None:38 self.db = db3940 def summary_statistics(41 self,42 panel: pd.DataFrame,43 anchor_ids: list[str],44 ) -> pd.DataFrame:45 cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns]46 stats = panel[cols].describe(percentiles=[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99]).T47 stats.columns = [48 "count", "mean", "std", "min", "p1", "p5", "p25",49 "median", "p75", "p95", "p99", "max",50 ]51 stats["skewness"] = panel[cols].skew()52 stats["kurtosis"] = panel[cols].kurtosis()53 stats.index = [c.replace("_mean", "") for c in cols]54 stats.index.name = "anchor"55 return stats5657 def correlation_matrix(58 self,59 panel: pd.DataFrame,60 anchor_ids: list[str],61 ) -> pd.DataFrame:62 cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns]63 corr = panel[cols].corr()64 corr.index = [c.replace("_mean", "") for c in cols]65 corr.columns = [c.replace("_mean", "") for c in cols]66 return corr6768 def top_firms(69 self,70 panel: pd.DataFrame,71 anchor_ids: list[str],72 n: int = 20,73 ) -> dict[str, pd.DataFrame]:74 results = {}75 for aid in anchor_ids:76 col = f"{aid}_mean"77 if col not in panel.columns:78 continue79 avg_score = (80 panel.groupby("ticker")[col]81 .mean()82 .sort_values(ascending=False)83 .head(n)84 .reset_index()85 )86 avg_score.columns = ["ticker", f"avg_{aid}"]87 companies = self.db.fetch_df(88 "SELECT ticker, company_name, sector FROM companies"89 )90 avg_score = avg_score.merge(companies, on="ticker", how="left")91 results[aid] = avg_score92 return results9394 def sector_averages(95 self,96 panel: pd.DataFrame,97 anchor_ids: list[str],98 ) -> pd.DataFrame:99 cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns]100 sector_avg = panel.groupby("sector")[cols].mean()101 sector_avg.columns = [c.replace("_mean", "") for c in sector_avg.columns]102 return sector_avg.sort_index()103104 def time_trends(105 self,106 panel: pd.DataFrame,107 anchor_ids: list[str],108 ) -> pd.DataFrame:109 cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns]110 trends = panel.groupby("fiscal_year")[cols].mean()111 trends.columns = [c.replace("_mean", "") for c in trends.columns]112 return trends.sort_index()113114 def coverage_report(self) -> dict:115 coverage = self.db.fetch_df("""116 SELECT117 fiscal_year,118 COUNT(DISTINCT ticker) AS firm_count,119 COUNT(DISTINCT accession_number) AS filing_count,120 SUM(chunk_count) AS total_chunks121 FROM filing_scores122 GROUP BY fiscal_year123 ORDER BY fiscal_year124 """)125 return {126 "by_year": coverage.to_dict("records"),127 "total_firms": int(coverage["firm_count"].max()),128 "total_years": len(coverage),129 "year_range": [int(coverage["fiscal_year"].min()), int(coverage["fiscal_year"].max())],130 }131132 def validate_scores(self, panel: pd.DataFrame, anchor_ids: list[str]) -> list[str]:133 warnings = []134 for aid in anchor_ids:135 col = f"{aid}_mean"136 if col not in panel.columns:137 warnings.append(f"Missing column: {col}")138 continue139 nan_count = panel[col].isna().sum()140 if nan_count > 0:141 warnings.append(f"{col}: {nan_count} NaN values")142 out_of_range = ((panel[col] < -1) | (panel[col] > 1)).sum()143 if out_of_range > 0:144 warnings.append(f"{col}: {out_of_range} values outside [-1, 1]")145 variance = panel[col].var()146 if variance < 1e-8:147 warnings.append(f"{col}: near-zero variance ({variance:.2e})")148 return warnings149150 def save_diagnostics(151 self,152 panel: pd.DataFrame,153 anchor_ids: list[str],154 output_dir: Path | None = None,155 ) -> None:156 out = output_dir or (PROJECT_ROOT / "outputs" / "diagnostics")157 out.mkdir(parents=True, exist_ok=True)158159 stats = self.summary_statistics(panel, anchor_ids)160 stats.to_csv(out / "summary_stats.csv")161 logger.info("Saved summary_stats.csv")162163 corr = self.correlation_matrix(panel, anchor_ids)164 corr.to_csv(out / "correlation_matrix.csv")165 logger.info("Saved correlation_matrix.csv")166167 top = self.top_firms(panel, anchor_ids)168 for aid, df in top.items():169 df.to_csv(out / f"top_firms_{aid}.csv", index=False)170 logger.info("Saved top_firms CSVs for %d anchors", len(top))171172 sector = self.sector_averages(panel, anchor_ids)173 sector.to_csv(out / "sector_averages.csv")174 logger.info("Saved sector_averages.csv")175176 trends = self.time_trends(panel, anchor_ids)177 trends.to_csv(out / "time_trends.csv")178 logger.info("Saved time_trends.csv")179180 coverage = self.coverage_report()181 with open(out / "coverage_report.json", "w") as f:182 json.dump(coverage, f, indent=2)183 logger.info("Saved coverage_report.json")184185 warnings = self.validate_scores(panel, anchor_ids)186 with open(out / "validation_warnings.txt", "w") as f:187 if warnings:188 f.write("\n".join(warnings))189 else:190 f.write("No validation warnings.\n")191 logger.info("Saved validation_warnings.txt (%d warnings)", len(warnings))192193 logger.info("All diagnostics saved to %s", out)194