# ============================================================ # Project: EDGAR Research Database # Description: SEC EDGAR Research Infrastructure # # Authors: # - Manel Kammoun — manel.kammoun@uqo.ca # - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca # - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca # # Institution: # Universite du Quebec en Outaouais (UQO) # # License: # Research / Academic Use # ============================================================ from __future__ import annotations import json from pathlib import Path from typing import TYPE_CHECKING import numpy as np import pandas as pd from src.utils.constants import PROJECT_ROOT from src.utils.logging import get_logger if TYPE_CHECKING: from src.storage.database import DatabaseManager logger = get_logger("factors.diagnostics") class FactorDiagnostics: def __init__(self, db: DatabaseManager) -> None: self.db = db def summary_statistics( self, panel: pd.DataFrame, anchor_ids: list[str], ) -> pd.DataFrame: cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns] stats = panel[cols].describe(percentiles=[0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99]).T stats.columns = [ "count", "mean", "std", "min", "p1", "p5", "p25", "median", "p75", "p95", "p99", "max", ] stats["skewness"] = panel[cols].skew() stats["kurtosis"] = panel[cols].kurtosis() stats.index = [c.replace("_mean", "") for c in cols] stats.index.name = "anchor" return stats def correlation_matrix( self, panel: pd.DataFrame, anchor_ids: list[str], ) -> pd.DataFrame: cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns] corr = panel[cols].corr() corr.index = [c.replace("_mean", "") for c in cols] corr.columns = [c.replace("_mean", "") for c in cols] return corr def top_firms( self, panel: pd.DataFrame, anchor_ids: list[str], n: int = 20, ) -> dict[str, pd.DataFrame]: results = {} for aid in anchor_ids: col = f"{aid}_mean" if col not in panel.columns: continue avg_score = ( panel.groupby("ticker")[col] .mean() .sort_values(ascending=False) .head(n) .reset_index() ) avg_score.columns = ["ticker", f"avg_{aid}"] companies = self.db.fetch_df( "SELECT ticker, company_name, sector FROM companies" ) avg_score = avg_score.merge(companies, on="ticker", how="left") results[aid] = avg_score return results def sector_averages( self, panel: pd.DataFrame, anchor_ids: list[str], ) -> pd.DataFrame: cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns] sector_avg = panel.groupby("sector")[cols].mean() sector_avg.columns = [c.replace("_mean", "") for c in sector_avg.columns] return sector_avg.sort_index() def time_trends( self, panel: pd.DataFrame, anchor_ids: list[str], ) -> pd.DataFrame: cols = [f"{aid}_mean" for aid in anchor_ids if f"{aid}_mean" in panel.columns] trends = panel.groupby("fiscal_year")[cols].mean() trends.columns = [c.replace("_mean", "") for c in trends.columns] return trends.sort_index() def coverage_report(self) -> dict: coverage = self.db.fetch_df(""" SELECT fiscal_year, COUNT(DISTINCT ticker) AS firm_count, COUNT(DISTINCT accession_number) AS filing_count, SUM(chunk_count) AS total_chunks FROM filing_scores GROUP BY fiscal_year ORDER BY fiscal_year """) return { "by_year": coverage.to_dict("records"), "total_firms": int(coverage["firm_count"].max()), "total_years": len(coverage), "year_range": [int(coverage["fiscal_year"].min()), int(coverage["fiscal_year"].max())], } def validate_scores(self, panel: pd.DataFrame, anchor_ids: list[str]) -> list[str]: warnings = [] for aid in anchor_ids: col = f"{aid}_mean" if col not in panel.columns: warnings.append(f"Missing column: {col}") continue nan_count = panel[col].isna().sum() if nan_count > 0: warnings.append(f"{col}: {nan_count} NaN values") out_of_range = ((panel[col] < -1) | (panel[col] > 1)).sum() if out_of_range > 0: warnings.append(f"{col}: {out_of_range} values outside [-1, 1]") variance = panel[col].var() if variance < 1e-8: warnings.append(f"{col}: near-zero variance ({variance:.2e})") return warnings def save_diagnostics( self, panel: pd.DataFrame, anchor_ids: list[str], output_dir: Path | None = None, ) -> None: out = output_dir or (PROJECT_ROOT / "outputs" / "diagnostics") out.mkdir(parents=True, exist_ok=True) stats = self.summary_statistics(panel, anchor_ids) stats.to_csv(out / "summary_stats.csv") logger.info("Saved summary_stats.csv") corr = self.correlation_matrix(panel, anchor_ids) corr.to_csv(out / "correlation_matrix.csv") logger.info("Saved correlation_matrix.csv") top = self.top_firms(panel, anchor_ids) for aid, df in top.items(): df.to_csv(out / f"top_firms_{aid}.csv", index=False) logger.info("Saved top_firms CSVs for %d anchors", len(top)) sector = self.sector_averages(panel, anchor_ids) sector.to_csv(out / "sector_averages.csv") logger.info("Saved sector_averages.csv") trends = self.time_trends(panel, anchor_ids) trends.to_csv(out / "time_trends.csv") logger.info("Saved time_trends.csv") coverage = self.coverage_report() with open(out / "coverage_report.json", "w") as f: json.dump(coverage, f, indent=2) logger.info("Saved coverage_report.json") warnings = self.validate_scores(panel, anchor_ids) with open(out / "validation_warnings.txt", "w") as f: if warnings: f.write("\n".join(warnings)) else: f.write("No validation warnings.\n") logger.info("Saved validation_warnings.txt (%d warnings)", len(warnings)) logger.info("All diagnostics saved to %s", out)