# ============================================================ # 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: # Université du Québec en Outaouais (UQO) # # License: # Research / Academic Use # ============================================================ from __future__ import annotations import argparse import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from src.embeddings import EmbeddingManager from src.search import SemanticSearch from src.storage.database import DatabaseManager from src.storage.parquet_manager import ParquetManager from src.utils.logging import setup_logging logger = setup_logging("search_filings") MAX_TEXT_DISPLAY = 200 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Semantic search across SEC filing content" ) parser.add_argument( "--query", type=str, required=True, help="Search query text", ) parser.add_argument( "--top-k", type=int, default=20, help="Number of results to return (default: 20)", ) parser.add_argument( "--form-type", type=str, default=None, help="Filter by form type (e.g., 10-K)", ) parser.add_argument( "--ticker", type=str, default=None, help="Filter by ticker symbol", ) parser.add_argument( "--start-date", type=str, default=None, help="Filter by filing start date (YYYY-MM-DD)", ) parser.add_argument( "--end-date", type=str, default=None, help="Filter by filing end date (YYYY-MM-DD)", ) return parser.parse_args() def format_results(results) -> str: if results.empty: return "No results found." lines: list[str] = [] lines.append("") lines.append(f"{'Rank':<6} {'Score':<8} {'Ticker':<8} {'Form':<8} {'Date':<12} {'Text Preview'}") lines.append("-" * 100) for i, row in results.iterrows(): text = str(row.get("text", ""))[:MAX_TEXT_DISPLAY].replace("\n", " ") if len(str(row.get("text", ""))) > MAX_TEXT_DISPLAY: text += "..." lines.append( f"{i + 1:<6} " f"{row.get('similarity_score', 0):<8.4f} " f"{str(row.get('ticker', '')):<8} " f"{str(row.get('form_type', '')):<8} " f"{str(row.get('filing_date', '')):<12} " f"{text}" ) lines.append("") lines.append(f"Total results: {len(results)}") return "\n".join(lines) def main() -> None: args = parse_args() logger.info("Query: '%s'", args.query) logger.info("Top-K: %d", args.top_k) db = DatabaseManager() db.initialize() parquet = ParquetManager() emb = EmbeddingManager(db, parquet) search = SemanticSearch(db, emb) try: results = search.search( query=args.query, top_k=args.top_k, form_type=args.form_type, ticker=args.ticker, start_date=args.start_date, end_date=args.end_date, ) except NotImplementedError as e: logger.error("Search failed: %s", e) db.close() sys.exit(1) except Exception: logger.error("Search failed", exc_info=True) db.close() sys.exit(1) print(format_results(results)) db.close() if __name__ == "__main__": main()