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# Université du Québec en Outaouais (UQO)12#13# License:14# Research / Academic Use15# ============================================================1617from __future__ import annotations1819import re20from pathlib import Path21from typing import TYPE_CHECKING2223import pandas as pd2425from src.utils.logging import get_logger2627if TYPE_CHECKING:28 from src.storage.database import DatabaseManager2930logger = get_logger("analytics.panel")313233class PanelBuilder:3435 def __init__(self, db: DatabaseManager) -> None:36 self.db = db3738 def build_panel(39 self,40 tickers: list[str] | None = None,41 form_types: list[str] | None = None,42 start_year: int | None = None,43 end_year: int | None = None,44 frequency: str = "annual",45 ) -> pd.DataFrame:46 conditions: list[str] = []47 params: list = []4849 if tickers:50 placeholders = ", ".join(["?"] * len(tickers))51 conditions.append(f"f.ticker IN ({placeholders})")52 params.extend(tickers)53 if form_types:54 placeholders = ", ".join(["?"] * len(form_types))55 conditions.append(f"f.form_type IN ({placeholders})")56 params.extend(form_types)57 if start_year is not None:58 conditions.append("f.fiscal_year >= ?")59 params.append(start_year)60 if end_year is not None:61 conditions.append("f.fiscal_year <= ?")62 params.append(end_year)6364 where = f" WHERE {' AND '.join(conditions)}" if conditions else ""6566 if frequency == "quarterly":67 group_cols = "f.ticker, c.company_name, f.cik, f.fiscal_year, f.fiscal_quarter, f.form_type"68 select_extra = "f.fiscal_quarter,"69 else:70 group_cols = "f.ticker, c.company_name, f.cik, f.fiscal_year, f.form_type"71 select_extra = ""7273 sql = f"""74 SELECT75 f.ticker,76 c.company_name,77 f.cik,78 f.fiscal_year,79 {select_extra}80 f.form_type,81 MAX(f.filing_date) AS filing_date,82 COALESCE(SUM(s.word_count), 0) AS word_count,83 COUNT(DISTINCT s.section_id) AS section_count,84 COALESCE(85 (SELECT COUNT(*) FROM filing_chunks ch WHERE ch.accession_number = MAX(f.accession_number)),86 087 ) AS chunk_count88 FROM filings f89 LEFT JOIN companies c ON f.cik = c.cik90 LEFT JOIN filing_sections s ON f.accession_number = s.accession_number91 {where}92 GROUP BY {group_cols}93 ORDER BY f.ticker, f.fiscal_year94 """9596 panel = self.db.fetch_df(sql, params if params else None)97 logger.info("Built panel with %d rows", len(panel))98 return panel99100 def add_section_stats(101 self,102 panel: pd.DataFrame,103 section_names: list[str] | None = None,104 ) -> pd.DataFrame:105 if panel.empty:106 return panel107108 tickers = panel["ticker"].unique().tolist()109 placeholders = ", ".join(["?"] * len(tickers))110111 sql = f"""112 SELECT113 f.ticker,114 f.fiscal_year,115 s.section_name,116 SUM(s.word_count) AS section_word_count117 FROM filing_sections s118 JOIN filings f ON s.accession_number = f.accession_number119 WHERE f.ticker IN ({placeholders})120 GROUP BY f.ticker, f.fiscal_year, s.section_name121 """122 section_df = self.db.fetch_df(sql, tickers)123124 if section_df.empty:125 return panel126127 if section_names:128 section_df = section_df[section_df["section_name"].isin(section_names)]129130 pivoted = section_df.pivot_table(131 index=["ticker", "fiscal_year"],132 columns="section_name",133 values="section_word_count",134 aggfunc="sum",135 fill_value=0,136 )137138 pivoted.columns = [self._sanitize_column_name(c) + "_words" for c in pivoted.columns]139 pivoted = pivoted.reset_index()140141 result = panel.merge(pivoted, on=["ticker", "fiscal_year"], how="left")142 for col in pivoted.columns:143 if col not in ("ticker", "fiscal_year"):144 result[col] = result[col].fillna(0).astype(int)145146 logger.info("Added %d section stat columns", len(pivoted.columns) - 2)147 return result148149 def add_custom_measure(150 self,151 panel: pd.DataFrame,152 measure_name: str,153 sql: str,154 ) -> pd.DataFrame:155 measure_df = self.db.fetch_df(sql)156157 if measure_df.empty:158 logger.warning("Custom measure '%s' returned no data", measure_name)159 return panel160161 merge_cols = [c for c in ["ticker", "fiscal_year", "cik"] if c in measure_df.columns and c in panel.columns]162 if not merge_cols:163 raise ValueError(164 f"Custom measure SQL must return at least one join column "165 f"(ticker, fiscal_year, or cik) matching the panel"166 )167168 result = panel.merge(measure_df, on=merge_cols, how="left")169 logger.info("Added custom measure '%s' via columns %s", measure_name, merge_cols)170 return result171172 def export_stata(self, panel: pd.DataFrame, output_path: Path) -> None:173 output_path = Path(output_path)174 output_path.parent.mkdir(parents=True, exist_ok=True)175176 df_clean = panel.copy()177 for col in df_clean.columns:178 if df_clean[col].dtype == "object":179 df_clean[col] = df_clean[col].astype(str).str[:244]180181 df_clean.columns = [self._sanitize_column_name(c) for c in df_clean.columns]182183 df_clean.to_stata(str(output_path), write_index=False, version=118)184 logger.info("Exported panel (%d rows) to Stata: %s", len(df_clean), output_path)185186 def export_csv(self, panel: pd.DataFrame, output_path: Path) -> None:187 output_path = Path(output_path)188 output_path.parent.mkdir(parents=True, exist_ok=True)189 panel.to_csv(output_path, index=False)190 logger.info("Exported panel (%d rows) to CSV: %s", len(panel), output_path)191192 @staticmethod193 def _sanitize_column_name(name: str) -> str:194 sanitized = re.sub(r"[^a-zA-Z0-9_]", "_", name.lower().strip())195 sanitized = re.sub(r"_+", "_", sanitized).strip("_")196 if sanitized and sanitized[0].isdigit():197 sanitized = "col_" + sanitized198 return sanitized[:32]199