"""Standalone research probe for SEC Form 4 idle-alpha overlays. This tool keeps all writes inside apps/tools and does not touch shared strategy manifests. It: 1. Replays a baseline manifest over a fixed date window. 2. Downloads SEC quarterly Form 4 flat files for the covered window. 3. Builds richer insider-buy cluster features than the original probe. 4. Scans standalone Form 4 specs, then tests the best candidates as an additive overlay versus the baseline's existing idle-alpha sleeve. The overlay model is intentionally conservative: - it only spends residual buying power already left idle by the baseline - it treats the displaced benchmark as the baseline parking sleeve - it skips unpublished SEC quarters instead of failing hard """ from __future__ import annotations import argparse import csv import datetime as dt import io import json import math import urllib.error import urllib.request import zipfile from collections import defaultdict from dataclasses import dataclass from pathlib import Path from typing import Any from apps.backtester.run import BacktestRunner, _build_merged_snapshot_store, load_manifest, resolve_config from libs.backtest.domain import DailyPortfolioState from libs.backtest.metrics import ( compute_max_drawdown_pct, compute_sharpe_ratio, compute_total_return_pct, ) from libs.common.logging import configure_logging _DEFAULT_USER_AGENT = "fithia2-form4-idle-alpha/1.0 (local research; contact: dev@example.com)" @dataclass(frozen=True) class RawForm4Transaction: symbol: str filing_date: dt.date transaction_date: dt.date owner_cik: str owner_relationship: str owner_title: str shares: float price: float total_value: float shares_owned_following: float purchase_pct_of_holding: float @dataclass(frozen=True) class Form4DailyEvent: symbol: str filing_date: dt.date total_value: float owner_count: int transaction_count: int event_day_count: int max_purchase_pct: float median_purchase_pct: float weighted_purchase_pct: float max_lag_days: int | None min_lag_days: int | None has_officer_or_director: bool @dataclass(frozen=True) class Form4Spec: name: str cluster_window_days: int min_owner_count: int min_total_value: float min_event_day_count: int min_purchase_pct: float max_lag_days: int | None hold_days: int max_positions: int = 6 max_new_per_day: int = 2 @dataclass(frozen=True) class BaselineContext: manifest_path: str start_date: dt.date signal_end_date: dt.date evaluation_end_date: dt.date initial_equity: float store: Any trading_days: list[dt.date] baseline_curve: list[DailyPortfolioState] baseline_equity_by_date: dict[dt.date, float] baseline_cash_by_date: dict[dt.date, float] baseline_metrics: dict[str, float] parking_symbol_by_date: dict[dt.date, str] def _parse_date(value: str, *, is_end: bool = False) -> dt.date: parts = value.split("-") if len(parts) == 1 and len(value) == 4 and value.isdigit(): year = int(value) return dt.date(year, 12, 31) if is_end else dt.date(year, 1, 1) if len(parts) == 2 and all(part.isdigit() for part in parts): year = int(parts[0]) month = int(parts[1]) if is_end: next_month = dt.date(year + (month // 12), (month % 12) + 1, 1) return next_month - dt.timedelta(days=1) return dt.date(year, month, 1) return dt.date.fromisoformat(value) def _quarter_range(start_date: dt.date, end_date: dt.date) -> list[tuple[int, int]]: year = start_date.year quarter = (start_date.month - 1) // 3 + 1 end_key = (end_date.year, (end_date.month - 1) // 3 + 1) quarters: list[tuple[int, int]] = [] while (year, quarter) <= end_key: quarters.append((year, quarter)) quarter += 1 if quarter == 5: year += 1 quarter = 1 return quarters def _quarter_zip_path(cache_dir: Path, year: int, quarter: int) -> Path: return cache_dir / f"{year}q{quarter}_form345.zip" def _quarter_zip_url(year: int, quarter: int) -> str: return ( "https://www.sec.gov/files/structureddata/data/" f"insider-transactions-data-sets/{year}q{quarter}_form345.zip" ) def _ensure_quarter_zip( cache_dir: Path, year: int, quarter: int, *, user_agent: str, ) -> Path | None: cache_dir.mkdir(parents=True, exist_ok=True) path = _quarter_zip_path(cache_dir, year, quarter) if path.exists() and path.stat().st_size > 0: return path request = urllib.request.Request( _quarter_zip_url(year, quarter), headers={"User-Agent": user_agent}, ) try: with urllib.request.urlopen(request, timeout=60) as response: data = response.read() except urllib.error.HTTPError as exc: if exc.code == 404: return None raise path.write_bytes(data) return path def _parse_sec_date(value: str | None) -> dt.date | None: if not value: return None try: return dt.datetime.strptime(value, "%d-%b-%Y").date() except ValueError: return None def _coerce_float(value: Any) -> float | None: try: result = float(value) except (TypeError, ValueError): return None if math.isnan(result) or math.isinf(result): return None return result def _normalize_title(text: str) -> str: normalized = text.strip().lower() return " ".join(normalized.split()) def _is_officer_or_director(relationship: str, title: str) -> bool: combined = f"{relationship} {title}".strip().lower() tokens = ( "director", "officer", "chief", "ceo", "cfo", "coo", "president", "chair", ) return any(token in combined for token in tokens) def _load_form4_transactions( *, cache_dir: Path, start_date: dt.date, end_date: dt.date, allowed_symbols: set[str], user_agent: str, ) -> tuple[list[RawForm4Transaction], list[str]]: transactions: list[RawForm4Transaction] = [] skipped_quarters: list[str] = [] for year, quarter in _quarter_range(start_date, end_date): path = _ensure_quarter_zip(cache_dir, year, quarter, user_agent=user_agent) if path is None: skipped_quarters.append(f"{year}Q{quarter}") continue with zipfile.ZipFile(path) as zf: submissions: dict[str, dict[str, Any]] = {} with zf.open("SUBMISSION.tsv") as handle: reader = csv.DictReader( io.TextIOWrapper(handle, encoding="utf-8", newline=""), delimiter="\t", ) for row in reader: symbol = str(row.get("ISSUERTRADINGSYMBOL") or "").strip().upper() if not symbol or symbol not in allowed_symbols: continue if str(row.get("DOCUMENT_TYPE") or "").strip().upper() != "4": continue filing_date = _parse_sec_date(row.get("FILING_DATE")) if filing_date is None or filing_date < start_date or filing_date > end_date: continue submissions[str(row["ACCESSION_NUMBER"])] = { "symbol": symbol, "filing_date": filing_date, } if not submissions: continue relationships: dict[str, list[tuple[str, str, str]]] = defaultdict(list) with zf.open("REPORTINGOWNER.tsv") as handle: reader = csv.DictReader( io.TextIOWrapper(handle, encoding="utf-8", newline=""), delimiter="\t", ) for row in reader: accession = str(row["ACCESSION_NUMBER"]) if accession not in submissions: continue owner_cik = str(row.get("RPTOWNERCIK") or "").strip() relationship = str(row.get("RPTOWNER_RELATIONSHIP") or "").strip().lower() title = _normalize_title(str(row.get("RPTOWNER_TITLE") or "")) relationships[accession].append((owner_cik, relationship, title)) with zf.open("NONDERIV_TRANS.tsv") as handle: reader = csv.DictReader( io.TextIOWrapper(handle, encoding="utf-8", newline=""), delimiter="\t", ) for row in reader: accession = str(row["ACCESSION_NUMBER"]) submission = submissions.get(accession) if submission is None: continue if str(row.get("TRANS_CODE") or "").strip().upper() != "P": continue if str(row.get("TRANS_ACQUIRED_DISP_CD") or "").strip().upper() != "A": continue shares = _coerce_float(row.get("TRANS_SHARES")) price = _coerce_float(row.get("TRANS_PRICEPERSHARE")) shares_following = _coerce_float(row.get("SHRS_OWND_FOLWNG_TRANS")) or 0.0 if shares is None or price is None or shares <= 0 or price <= 0: continue transaction_date = ( _parse_sec_date(row.get("TRANS_DATE")) or submission["filing_date"] ) purchase_pct = ( shares / shares_following if shares_following > 0 else 0.0 ) owner_rows = relationships.get(accession) or [("", "", "")] for owner_cik, relationship, title in owner_rows: transactions.append( RawForm4Transaction( symbol=submission["symbol"], filing_date=submission["filing_date"], transaction_date=transaction_date, owner_cik=owner_cik, owner_relationship=relationship, owner_title=title, shares=shares, price=price, total_value=shares * price, shares_owned_following=shares_following, purchase_pct_of_holding=purchase_pct, ) ) return transactions, skipped_quarters def _aggregate_daily_events(transactions: list[RawForm4Transaction]) -> list[Form4DailyEvent]: grouped: dict[tuple[dt.date, str], dict[str, Any]] = {} for row in transactions: key = (row.filing_date, row.symbol) event = grouped.setdefault( key, { "filing_date": row.filing_date, "symbol": row.symbol, "total_value": 0.0, "transaction_count": 0, "owner_ciks": set(), "purchase_pcts": [], "weighted_num": 0.0, "weighted_den": 0.0, "lag_days": [], "has_officer_or_director": False, }, ) event["total_value"] += row.total_value event["transaction_count"] += 1 if row.owner_cik: event["owner_ciks"].add(row.owner_cik) if row.purchase_pct_of_holding > 0: event["purchase_pcts"].append(row.purchase_pct_of_holding) if row.shares_owned_following > 0: event["weighted_num"] += row.shares event["weighted_den"] += row.shares_owned_following lag_days = (row.filing_date - row.transaction_date).days event["lag_days"].append(lag_days) if _is_officer_or_director(row.owner_relationship, row.owner_title): event["has_officer_or_director"] = True events: list[Form4DailyEvent] = [] for payload in grouped.values(): purchase_pcts = payload["purchase_pcts"] or [0.0] weighted_purchase_pct = ( payload["weighted_num"] / payload["weighted_den"] if payload["weighted_den"] > 0 else 0.0 ) lag_days = payload["lag_days"] events.append( Form4DailyEvent( symbol=str(payload["symbol"]), filing_date=payload["filing_date"], total_value=float(payload["total_value"]), owner_count=len(payload["owner_ciks"]), transaction_count=int(payload["transaction_count"]), event_day_count=1, max_purchase_pct=max(purchase_pcts), median_purchase_pct=sorted(purchase_pcts)[len(purchase_pcts) // 2], weighted_purchase_pct=weighted_purchase_pct, max_lag_days=max(lag_days) if lag_days else None, min_lag_days=min(lag_days) if lag_days else None, has_officer_or_director=bool(payload["has_officer_or_director"]), ) ) events.sort(key=lambda row: (row.symbol, row.filing_date)) return events def _build_cluster_events( daily_events: list[Form4DailyEvent], *, window_days: int, ) -> list[Form4DailyEvent]: if window_days <= 0: return daily_events by_symbol: dict[str, list[Form4DailyEvent]] = defaultdict(list) for event in daily_events: by_symbol[event.symbol].append(event) clusters: list[Form4DailyEvent] = [] for symbol, events in by_symbol.items(): events.sort(key=lambda row: row.filing_date) left = 0 owner_counts: dict[int, int] = defaultdict(int) rolling_total_value = 0.0 rolling_transactions = 0 rolling_has_role = False for right, event in enumerate(events): cutoff = event.filing_date - dt.timedelta(days=window_days) while left <= right and events[left].filing_date < cutoff: old = events[left] rolling_total_value -= old.total_value rolling_transactions -= old.transaction_count left += 1 rolling_total_value += event.total_value rolling_transactions += event.transaction_count cluster_items = events[left : right + 1] rolling_has_role = any(item.has_officer_or_director for item in cluster_items) purchase_values = [item.max_purchase_pct for item in cluster_items] median_values = [item.median_purchase_pct for item in cluster_items] weighted_values = [item.weighted_purchase_pct for item in cluster_items] lag_values = [item.max_lag_days for item in cluster_items if item.max_lag_days is not None] owner_set: set[str] = set() for item in cluster_items: # owner_count is already deduped within the filing day, but we need # a proxy across days without carrying the full owner identity. # Use the observed owner_count sum capped by transaction_count when # cross-day owner identity is unavailable after aggregation. owner_set.update({f"{item.symbol}:{item.filing_date}:{idx}" for idx in range(item.owner_count)}) clusters.append( Form4DailyEvent( symbol=symbol, filing_date=event.filing_date, total_value=rolling_total_value, owner_count=len(owner_set), transaction_count=rolling_transactions, event_day_count=len(cluster_items), max_purchase_pct=max(purchase_values) if purchase_values else 0.0, median_purchase_pct=sorted(median_values)[len(median_values) // 2] if median_values else 0.0, weighted_purchase_pct=max(weighted_values) if weighted_values else 0.0, max_lag_days=max(lag_values) if lag_values else None, min_lag_days=min(lag_values) if lag_values else None, has_officer_or_director=rolling_has_role, ) ) clusters.sort(key=lambda row: (row.symbol, row.filing_date)) return clusters def _build_entries_for_spec( *, store: Any, events: list[Form4DailyEvent], spec: Form4Spec, ) -> dict[dt.date, list[dict[str, Any]]]: trading_days = store.all_trading_days() trading_index = {date: idx for idx, date in enumerate(trading_days)} entries: dict[dt.date, list[dict[str, Any]]] = defaultdict(list) for event in events: if event.owner_count < spec.min_owner_count: continue if event.total_value < spec.min_total_value: continue if event.event_day_count < spec.min_event_day_count: continue if event.weighted_purchase_pct < spec.min_purchase_pct: continue if spec.max_lag_days is not None and event.max_lag_days is not None and event.max_lag_days > spec.max_lag_days: continue entry_date = next((date for date in trading_days if date > event.filing_date), None) if entry_date is None: continue entry_bar = store.get_bar(event.symbol, entry_date) if not entry_bar or not entry_bar.get("open") or not entry_bar.get("close"): continue entry_index = trading_index.get(entry_date) if entry_index is None or entry_index + spec.hold_days >= len(trading_days): continue exit_date = trading_days[entry_index + spec.hold_days] exit_bar = store.get_bar(event.symbol, exit_date) if not exit_bar or not exit_bar.get("close"): continue entries[entry_date].append( { "symbol": event.symbol, "exit_date": exit_date, "score": ( event.owner_count, event.event_day_count, round(event.weighted_purchase_pct, 6), round(event.total_value, 2), ), "owner_count": event.owner_count, "event_day_count": event.event_day_count, "total_value": event.total_value, "purchase_pct": event.weighted_purchase_pct, } ) return entries def _dedupe_curve_by_date(curve: list[DailyPortfolioState]) -> list[DailyPortfolioState]: by_date: dict[dt.date, DailyPortfolioState] = {} for state in curve: by_date[state.date] = state return [by_date[date] for date in sorted(by_date)] def _build_curve_metrics(curve: list[DailyPortfolioState]) -> dict[str, float]: return { "total_return_pct": round(compute_total_return_pct(curve) or 0.0, 2), "max_drawdown_pct": round(compute_max_drawdown_pct(curve) or 0.0, 2), "sharpe_ratio": round(compute_sharpe_ratio(curve) or 0.0, 3), } def _yearly_return_map(curve: list[DailyPortfolioState]) -> dict[str, float]: if not curve: return {} year_start: dict[int, float] = {} year_end: dict[int, float] = {} for state in curve: year_start.setdefault(state.date.year, state.equity) year_end[state.date.year] = state.equity return { str(year): round((year_end[year] / year_start[year] - 1.0) * 100.0, 2) for year in sorted(year_start) if year_start[year] > 0 } def _simulate_equal_weight_curve( *, store: Any, start_date: dt.date, end_date: dt.date, entries_by_date: dict[dt.date, list[dict[str, Any]]], spec: Form4Spec, capital: float, ) -> tuple[list[DailyPortfolioState], int]: trading_days = [ date for date in store.all_trading_days() if start_date <= date <= end_date ] equity = capital peak = capital curve: list[DailyPortfolioState] = [] open_positions: list[dict[str, Any]] = [] trade_count = 0 for date in trading_days: if date in entries_by_date and len(open_positions) < spec.max_positions: existing_symbols = {position["symbol"] for position in open_positions} ranked = sorted(entries_by_date[date], key=lambda row: row["score"], reverse=True) added = 0 for row in ranked: if row["symbol"] in existing_symbols: continue bar = store.get_bar(row["symbol"], date) if not bar or not bar.get("open"): continue open_positions.append( { "symbol": row["symbol"], "exit_date": row["exit_date"], "prev_price": float(bar["open"]), } ) existing_symbols.add(row["symbol"]) trade_count += 1 added += 1 if added >= spec.max_new_per_day or len(open_positions) >= spec.max_positions: break if open_positions: daily_returns: list[float] = [] updated_positions: list[dict[str, Any]] = [] for position in open_positions: bar = store.get_bar(position["symbol"], date) if not bar or not bar.get("close"): continue close_price = float(bar["close"]) prev_price = float(position["prev_price"]) if prev_price <= 0: continue daily_returns.append(close_price / prev_price - 1.0) updated_positions.append( { "symbol": position["symbol"], "exit_date": position["exit_date"], "prev_price": close_price, } ) if daily_returns: equity *= 1.0 + sum(daily_returns) / len(daily_returns) open_positions = [ position for position in updated_positions if position["exit_date"] > date ] peak = max(peak, equity) curve.append( DailyPortfolioState( date=date, equity=equity, sizing_equity=equity, cash_available=equity, gross_exposure=float(len(open_positions)) * 10.0, net_exposure=float(len(open_positions)) * 10.0, reserved_risk_budget=0.0, unrealized_pnl=0.0, realized_pnl=0.0, open_positions=[position["symbol"] for position in open_positions], daily_new_risk_used=0.0, peak_equity=peak, current_drawdown_pct=((peak - equity) / peak * 100.0) if peak > 0 else 0.0, ) ) return curve, trade_count def _build_parking_symbol_by_date(context: BaselineContext, config: Any) -> dict[dt.date, str]: store = context.store runner = BacktestRunner( manifest=load_manifest(context.manifest_path), config=config, store=store, initial_equity=context.initial_equity, enable_engine_analysis=False, ) trading_days = context.trading_days if not trading_days: return {} held_symbol = runner._evaluate_parking_target(trading_days[0]) or "sgov" runner._commit_parking_target(held_symbol) parking_symbol_by_date: dict[dt.date, str] = {trading_days[0]: held_symbol} for idx in range(1, len(trading_days)): date = trading_days[idx] parking_symbol_by_date[date] = held_symbol next_symbol = runner._evaluate_parking_target(date) or held_symbol runner._commit_parking_target(next_symbol) held_symbol = next_symbol return parking_symbol_by_date def _build_baseline_context( *, manifest_path: str, start_date: dt.date, end_date: dt.date, initial_equity: float, ) -> BaselineContext: manifest = load_manifest(manifest_path) config = resolve_config(manifest) store = _build_merged_snapshot_store( manifest, config, snapshot_dir_override=None, ).slice_by_date_range(start_date, end_date) runner = BacktestRunner( manifest=manifest, config=config, store=store, initial_equity=initial_equity, enable_engine_analysis=False, ) result = runner.run(output_root=None) baseline_curve = _dedupe_curve_by_date(runner._equity_curve) trading_days = [state.date for state in baseline_curve] baseline_equity_by_date = {state.date: float(state.equity) for state in baseline_curve} baseline_cash_by_date = {state.date: float(state.cash_available) for state in baseline_curve} baseline_metrics = _build_curve_metrics(baseline_curve) baseline_context = BaselineContext( manifest_path=manifest_path, start_date=start_date, signal_end_date=end_date, evaluation_end_date=trading_days[-1] if trading_days else end_date, initial_equity=initial_equity, store=store, trading_days=trading_days, baseline_curve=baseline_curve, baseline_equity_by_date=baseline_equity_by_date, baseline_cash_by_date=baseline_cash_by_date, baseline_metrics=baseline_metrics, parking_symbol_by_date={}, ) parking_symbol_by_date = _build_parking_symbol_by_date(baseline_context, config) return BaselineContext( manifest_path=manifest_path, start_date=start_date, signal_end_date=end_date, evaluation_end_date=trading_days[-1] if trading_days else end_date, initial_equity=initial_equity, store=store, trading_days=trading_days, baseline_curve=baseline_curve, baseline_equity_by_date=baseline_equity_by_date, baseline_cash_by_date=baseline_cash_by_date, baseline_metrics=baseline_metrics, parking_symbol_by_date=parking_symbol_by_date, ) def _simulate_overlay_curve( *, context: BaselineContext, entries_by_date: dict[dt.date, list[dict[str, Any]]], spec: Form4Spec, deploy_idle_fraction: float, ) -> tuple[list[DailyPortfolioState], int]: store = context.store trading_days = context.trading_days open_positions: list[dict[str, Any]] = [] realized_excess = 0.0 trade_count = 0 curve: list[DailyPortfolioState] = [] peak = context.initial_equity for date_index, date in enumerate(trading_days): benchmark_symbol = context.parking_symbol_by_date.get(date, "sgov").upper() benchmark_bar = store.get_bar(benchmark_symbol, date) if benchmark_bar is None or benchmark_bar.get("close") is None: benchmark_bar = store.get_bar("SGOV", date) updated_positions: list[dict[str, Any]] = [] for position in open_positions: bar = store.get_bar(position["symbol"], date) if not bar or not bar.get("close"): continue actual_close = float(bar["close"]) shadow_close = float((benchmark_bar or {}).get("close") or position["shadow_prev_close"]) actual_value = position["actual_value"] shadow_value = position["shadow_value"] if position["prev_actual_price"] > 0: actual_value *= actual_close / position["prev_actual_price"] if position["shadow_prev_close"] > 0: shadow_value *= shadow_close / position["shadow_prev_close"] next_state = { **position, "actual_value": actual_value, "shadow_value": shadow_value, "prev_actual_price": actual_close, "shadow_prev_close": shadow_close, } if position["exit_date"] <= date: realized_excess += actual_value - shadow_value else: updated_positions.append(next_state) open_positions = updated_positions baseline_equity = context.baseline_equity_by_date[date] baseline_cash = context.baseline_cash_by_date[date] shadow_notional_open = sum(position["shadow_value"] for position in open_positions) idle_budget = max(0.0, baseline_cash * deploy_idle_fraction - shadow_notional_open) if date in entries_by_date and len(open_positions) < spec.max_positions and idle_budget > 0: existing_symbols = {position["symbol"] for position in open_positions} ranked = sorted(entries_by_date[date], key=lambda row: row["score"], reverse=True) remaining_slots = max(0, spec.max_positions - len(open_positions)) ranked = [ row for row in ranked if row["symbol"] not in existing_symbols ][: min(spec.max_new_per_day, remaining_slots)] if ranked: per_position_budget = idle_budget / len(ranked) benchmark_open = float((benchmark_bar or {}).get("open") or (benchmark_bar or {}).get("close") or 0.0) benchmark_close = float((benchmark_bar or {}).get("close") or benchmark_open or 0.0) for row in ranked: bar = store.get_bar(row["symbol"], date) if not bar or not bar.get("open") or not bar.get("close"): continue actual_open = float(bar["open"]) actual_close = float(bar["close"]) if per_position_budget <= 0 or actual_open <= 0 or benchmark_open <= 0: continue open_positions.append( { "symbol": row["symbol"], "exit_date": row["exit_date"], "actual_value": per_position_budget * (actual_close / actual_open), "shadow_value": per_position_budget * (benchmark_close / benchmark_open), "prev_actual_price": actual_close, "shadow_prev_close": benchmark_close, } ) trade_count += 1 open_excess = sum(position["actual_value"] - position["shadow_value"] for position in open_positions) equity = baseline_equity + realized_excess + open_excess peak = max(peak, equity) curve.append( DailyPortfolioState( date=date, equity=equity, sizing_equity=equity, cash_available=max(0.0, context.baseline_cash_by_date[date] - sum(position["shadow_value"] for position in open_positions)), gross_exposure=float(len(open_positions)) * 10.0, net_exposure=float(len(open_positions)) * 10.0, reserved_risk_budget=0.0, unrealized_pnl=open_excess, realized_pnl=realized_excess, open_positions=[position["symbol"] for position in open_positions], daily_new_risk_used=0.0, peak_equity=peak, current_drawdown_pct=((peak - equity) / peak * 100.0) if peak > 0 else 0.0, ) ) return curve, trade_count def _standalone_score(metrics: dict[str, float]) -> float: drawdown = max(metrics["max_drawdown_pct"], 1.0) return metrics["total_return_pct"] / drawdown + metrics["sharpe_ratio"] * 25.0 def _overlay_score(metrics: dict[str, float], baseline_metrics: dict[str, float]) -> float: delta_return = metrics["total_return_pct"] - baseline_metrics["total_return_pct"] delta_drawdown = metrics["max_drawdown_pct"] - baseline_metrics["max_drawdown_pct"] return delta_return * 4.0 - max(delta_drawdown, 0.0) * 35.0 + metrics["sharpe_ratio"] * 10.0 def _generate_specs() -> list[Form4Spec]: specs: list[Form4Spec] = [] for cluster_window_days in (0, 3, 5, 10): for min_owner_count in (2, 3): for min_total_value in (1_000_000.0, 2_000_000.0, 5_000_000.0): for min_event_day_count in ((1,) if cluster_window_days == 0 else (1, 2)): for min_purchase_pct in (0.0, 0.03, 0.05): for max_lag_days in (None, 2, 4): for hold_days in (5, 10, 20): name = ( f"w{cluster_window_days}_o{min_owner_count}" f"_v{int(min_total_value/1_000_000)}m" f"_d{min_event_day_count}" f"_p{int(min_purchase_pct*100):02d}" f"_lag{max_lag_days if max_lag_days is not None else 'na'}" f"_h{hold_days}" ) specs.append( Form4Spec( name=name, cluster_window_days=cluster_window_days, min_owner_count=min_owner_count, min_total_value=min_total_value, min_event_day_count=min_event_day_count, min_purchase_pct=min_purchase_pct, max_lag_days=max_lag_days, hold_days=hold_days, ) ) return specs def _run_probe(args: argparse.Namespace) -> dict[str, Any]: start_date = _parse_date(args.start) end_date = _parse_date(args.end, is_end=True) baseline = _build_baseline_context( manifest_path=args.config, start_date=start_date, end_date=end_date, initial_equity=args.capital, ) allowed_symbols = {str(symbol).upper() for symbol in baseline.store._bars.keys()} transactions, skipped_quarters = _load_form4_transactions( cache_dir=Path(args.cache_dir), start_date=start_date, end_date=end_date, allowed_symbols=allowed_symbols, user_agent=args.user_agent, ) daily_events = _aggregate_daily_events(transactions) events_by_window: dict[int, list[Form4DailyEvent]] = {} specs = _generate_specs() standalone_rows: list[dict[str, Any]] = [] for spec in specs: if spec.cluster_window_days not in events_by_window: events_by_window[spec.cluster_window_days] = _build_cluster_events( daily_events, window_days=spec.cluster_window_days, ) entries = _build_entries_for_spec( store=baseline.store, events=events_by_window[spec.cluster_window_days], spec=spec, ) curve, trade_count = _simulate_equal_weight_curve( store=baseline.store, start_date=start_date, end_date=end_date, entries_by_date=entries, spec=spec, capital=args.capital, ) metrics = _build_curve_metrics(curve) standalone_rows.append( { "spec": spec.name, "cluster_window_days": spec.cluster_window_days, "signals": int(sum(len(value) for value in entries.values())), "trades": trade_count, **metrics, "standalone_score": round(_standalone_score(metrics), 3), "yearly_return_pct": _yearly_return_map(curve), } ) standalone_rows.sort(key=lambda row: row["standalone_score"], reverse=True) spec_lookup = {spec.name: spec for spec in specs} overlay_rows: list[dict[str, Any]] = [] deploy_idle_fractions = (0.03, 0.04, 0.05, 0.06, 0.08, 0.1) for spec_name, spec in spec_lookup.items(): entries = _build_entries_for_spec( store=baseline.store, events=events_by_window[spec.cluster_window_days], spec=spec, ) for deploy_idle_fraction in deploy_idle_fractions: curve, trade_count = _simulate_overlay_curve( context=baseline, entries_by_date=entries, spec=spec, deploy_idle_fraction=deploy_idle_fraction, ) metrics = _build_curve_metrics(curve) overlay_rows.append( { "spec": spec_name, "deploy_idle_fraction": deploy_idle_fraction, "signals": int(sum(len(value) for value in entries.values())), "trades": trade_count, **metrics, "delta_return_pct": round(metrics["total_return_pct"] - baseline.baseline_metrics["total_return_pct"], 2), "delta_drawdown_pct": round(metrics["max_drawdown_pct"] - baseline.baseline_metrics["max_drawdown_pct"], 2), "overlay_score": round(_overlay_score(metrics, baseline.baseline_metrics), 3), "yearly_return_pct": _yearly_return_map(curve), } ) overlay_rows.sort(key=lambda row: row["overlay_score"], reverse=True) low_dd_overlay_rows = [ row for row in overlay_rows if row["delta_drawdown_pct"] <= 0.5 ] low_dd_overlay_rows.sort( key=lambda row: (row["delta_return_pct"], row["sharpe_ratio"]), reverse=True, ) best_overlay = low_dd_overlay_rows[0] if low_dd_overlay_rows else (overlay_rows[0] if overlay_rows else None) verdict = "not_tested" if best_overlay is not None: if best_overlay["delta_return_pct"] >= 25.0 and best_overlay["delta_drawdown_pct"] <= 0.5: verdict = "worth_integrating" elif best_overlay["delta_return_pct"] > 0: verdict = "maybe_research_more" else: verdict = "not_worth_integrating" return { "window": { "start_date": start_date.isoformat(), "signal_end_date": end_date.isoformat(), "evaluation_end_date": baseline.evaluation_end_date.isoformat(), "skipped_quarters": skipped_quarters, }, "baseline": { "manifest": args.config, **baseline.baseline_metrics, "yearly_return_pct": _yearly_return_map(baseline.baseline_curve), }, "form4_data": { "transactions": len(transactions), "daily_events": len(daily_events), "cluster_windows_tested": sorted(events_by_window), }, "standalone_top": standalone_rows[: args.report_top_n], "overlay_top": overlay_rows[: args.report_top_n], "overlay_low_dd_top": low_dd_overlay_rows[: args.report_top_n], "verdict": verdict, } def main() -> None: parser = argparse.ArgumentParser(description="Probe SEC Form 4 idle-alpha overlays") parser.add_argument("--config", required=True, help="Baseline experiment manifest JSON") parser.add_argument("--start", default="2022-03-03") parser.add_argument("--end", default="2025-12-31") parser.add_argument("--capital", type=float, default=10_000.0) parser.add_argument("--cache-dir", default="data/cache/sec_form345") parser.add_argument("--user-agent", default=_DEFAULT_USER_AGENT) parser.add_argument("--overlay-top-k", type=int, default=12) parser.add_argument("--report-top-n", type=int, default=8) parser.add_argument("--json", action="store_true") args = parser.parse_args() configure_logging("WARNING") result = _run_probe(args) if args.json: print(json.dumps(result, indent=2)) else: print(result) if __name__ == "__main__": main()