From 4d0e773ba0c0e5108b40ac7ddc82aad3bd17a208 Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Fri, 13 Mar 2026 20:04:26 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20overhaul=20strategy=20=E2=80=94=20docum?= =?UTF-8?q?ent=20quality=20>=20price=20momentum?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Flip scoring weights so event/document quality is primary signal (55%) and market confirmation is secondary (35%). Add research mode with kill-switch cooldown/reset, veto gates for bad events, reduced portfolio risk, and 4 diagnostic analysis scripts. Phase A: Research mode kill-switch reset, risk reduction (0.5%/trade, max 4 positions), bullish-only direction for all event types. Phase B: 2 new sub-scorers (parse_confidence, direction_clarity), 4 veto gates (oneoff risk, parse confidence, unknown/bearish direction). Phase C: signal_quality, event_type_decomposition, kill_switch_impact, concurrent_position analysis scripts. Co-Authored-By: Claude Opus 4.6 --- apps/backtester/run.py | 208 +++++++++++++++++-- configs/backtest/defaults.json | 24 ++- dev/analysis/concurrent_position_analysis.py | 169 +++++++++++++++ dev/analysis/event_type_decomposition.py | 122 +++++++++++ dev/analysis/kill_switch_impact_analysis.py | 122 +++++++++++ dev/analysis/signal_quality_analysis.py | 170 +++++++++++++++ libs/backtest/allocator.py | 29 ++- libs/backtest/domain.py | 11 + libs/backtest/scoring.py | 142 ++++++++++--- tests/unit/backtest/test_allocator.py | 78 +++++++ tests/unit/backtest/test_scoring.py | 52 ++++- 11 files changed, 1069 insertions(+), 58 deletions(-) create mode 100644 dev/analysis/concurrent_position_analysis.py create mode 100644 dev/analysis/event_type_decomposition.py create mode 100644 dev/analysis/kill_switch_impact_analysis.py create mode 100644 dev/analysis/signal_quality_analysis.py diff --git a/apps/backtester/run.py b/apps/backtester/run.py index 9cd68e4..9b0aac7 100644 --- a/apps/backtester/run.py +++ b/apps/backtester/run.py @@ -3,6 +3,7 @@ from __future__ import annotations import argparse import datetime as dt +import json import subprocess import sys from pathlib import Path @@ -31,6 +32,7 @@ from libs.backtest.manifests import generate_run_id, load_manifest, resolve_conf from libs.backtest.metrics import build_metrics_bundle from libs.backtest.selector import select_candidates from libs.backtest.snapshot_store import SnapshotStore +from libs.backtest.splits import generate_walk_forward_windows from libs.common.logging import get_logger from libs.common.time_utils import utc_now @@ -82,6 +84,7 @@ class BacktestRunner: self._consecutive_losses = 0 self._cooldown_remaining = 0 self._kill_switch_triggered = False + self._kill_switch_cooldown_remaining = 0 def run(self, output_root: str | Path | None = None) -> ExperimentResult: """Execute the full simulation. Returns ExperimentResult.""" @@ -178,9 +181,11 @@ class BacktestRunner: # Reset daily risk tracker self._daily_new_risk_used = 0.0 - # Decrement cooldown + # Decrement cooldowns if self._cooldown_remaining > 0: self._cooldown_remaining -= 1 + if self._kill_switch_cooldown_remaining > 0: + self._kill_switch_cooldown_remaining -= 1 # Increment days_held for all open positions for pos in self._open_positions: @@ -261,6 +266,18 @@ class BacktestRunner: if drawdown_pct >= _KILL_SWITCH_DRAWDOWN_PCT and not self._kill_switch_triggered: logger.warning("kill_switch_triggered", date=str(date), drawdown_pct=drawdown_pct) self._kill_switch_triggered = True + if self.config.risk.backtest_mode == "research": + self._kill_switch_cooldown_remaining = self.config.risk.kill_switch_cooldown_days + + # Research mode: reset kill switch after cooldown + recovery + if ( + self._kill_switch_triggered + and self.config.risk.backtest_mode == "research" + and self._kill_switch_cooldown_remaining <= 0 + and drawdown_pct < _KILL_SWITCH_DRAWDOWN_PCT + ): + self._kill_switch_triggered = False + logger.info("kill_switch_reset", date=str(date), drawdown_pct=drawdown_pct) # --- ENTRIES (only if kill switch not triggered) --- if not self._kill_switch_triggered: @@ -424,6 +441,143 @@ def _build_store( ) +def run_walk_forward( + manifest: ExperimentManifest, + config: BacktestConfig, + snapshot_dir_override: str | None, + initial_equity: float, + output_root: str, + train_days: int = 252, + test_days: int = 63, + step_days: int | None = None, +) -> list[dict[str, Any]]: + """Run walk-forward cross-validation over all splits. + + Loads all three splits (train/valid/test) into one SnapshotStore, + generates walk-forward windows, and runs a separate backtest on each + window's test period. Returns per-fold metrics. + """ + from libs.common.config import get_settings + + s = get_settings() + snapshot_root = Path(snapshot_dir_override or s.parquet_dir) / config.dataset_snapshot_id + + # Merge all splits into a single SnapshotStore + stores: list[SnapshotStore] = [] + for split in ["train", "valid", "test"]: + parquet_path = snapshot_root / f"{split}.parquet" + if parquet_path.exists(): + st = SnapshotStore.load( + snapshot_dir=snapshot_root, + split_name=split, + oracle_url=s.stock_oracle_url, + db_dsn=s.postgres_dsn, + ) + stores.append(st) + + if not stores: + print("No splits found to load.") + return [] + + # Merge candidates and bars from all stores + merged_candidates: dict[dt.date, list[dict[str, Any]]] = {} + merged_bars: dict[str, dict[dt.date, dict[str, Any]]] = {} + merged_macro: dict[dt.date, dict[str, Any]] = {} + + for st in stores: + for d in st.all_execution_dates(): + merged_candidates.setdefault(d, []).extend(st.get_candidates_for_date(d)) + merged_bars.update(st._bars) + merged_macro.update(st._macro) + + merged_store = SnapshotStore( + candidates_by_exec_date=merged_candidates, + bars_by_symbol_date=merged_bars, + macro_by_date=merged_macro, + ) + + all_dates = merged_store.all_trading_days() + if not all_dates: + print("No trading days found in merged data.") + return [] + + windows = generate_walk_forward_windows( + all_dates, + train_days=train_days, + test_days=test_days, + step_days=step_days, + ) + + if not windows: + print(f"Not enough data for walk-forward windows (need {train_days + test_days} days, have {len(all_dates)}).") + # Fall back to a single window using all available data + print("Running single-window backtest on all data instead.") + windows_dates = [(all_dates[0], all_dates[-1])] + else: + windows_dates = [(w.test_start, w.test_end) for w in windows] + + fold_results: list[dict[str, Any]] = [] + + for fold_idx, (test_start, test_end) in enumerate(windows_dates): + # Filter candidates to only those within the test window + filtered_candidates: dict[dt.date, list[dict[str, Any]]] = {} + for d, cands in merged_candidates.items(): + if test_start <= d <= test_end: + filtered_candidates[d] = cands + + fold_store = SnapshotStore( + candidates_by_exec_date=filtered_candidates, + bars_by_symbol_date=merged_bars, + macro_by_date=merged_macro, + ) + + runner = BacktestRunner( + manifest=manifest, + config=config, + store=fold_store, + initial_equity=initial_equity, + ) + result = runner.run(output_root=output_root) + + fold_info = { + "fold": fold_idx, + "test_start": str(test_start), + "test_end": str(test_end), + "trade_count": result.metrics.trade_count, + "total_return_pct": result.metrics.total_return_pct, + "win_rate": result.metrics.win_rate, + "profit_factor": result.metrics.profit_factor, + "max_drawdown_pct": result.metrics.max_drawdown_pct, + "sharpe_ratio": result.metrics.sharpe_ratio, + "run_id": result.run_id, + } + fold_results.append(fold_info) + print( + f"Fold {fold_idx}: {test_start}→{test_end} | " + f"Trades={result.metrics.trade_count} " + f"Return={result.metrics.total_return_pct or 0:.2f}% " + f"WinRate={result.metrics.win_rate or 0:.1%}" + ) + + # Aggregate summary + total_trades = sum(f["trade_count"] or 0 for f in fold_results) + returns = [f["total_return_pct"] for f in fold_results if f["total_return_pct"] is not None] + win_rates = [f["win_rate"] for f in fold_results if f["win_rate"] is not None] + + print(f"\n--- Walk-Forward Summary ({len(fold_results)} folds) ---") + print(f"Total trades: {total_trades}") + if returns: + import statistics + print(f"Mean return: {statistics.mean(returns):.2f}%") + if len(returns) > 1: + print(f"StdDev return: {statistics.stdev(returns):.2f}%") + if win_rates: + import statistics + print(f"Mean win rate: {statistics.mean(win_rates):.1%}") + + return fold_results + + def main() -> None: parser = argparse.ArgumentParser(description="ACE-F Backtester") parser.add_argument("--manifest", required=True, help="Path to experiment manifest JSON") @@ -433,23 +587,51 @@ def main() -> None: parser.add_argument("--output-root", default="./runs", help="Output root directory") parser.add_argument("--initial-equity", type=float, default=100_000.0) parser.add_argument("--config-root", default=".", help="Root dir for resolving config paths") + parser.add_argument( + "--walk-forward", + action="store_true", + help="Run walk-forward cross-validation instead of single backtest", + ) + parser.add_argument("--wf-train-days", type=int, default=252, help="Walk-forward train window (trading days)") + parser.add_argument("--wf-test-days", type=int, default=63, help="Walk-forward test window (trading days)") + parser.add_argument("--mode", choices=["research", "live"], default=None, + help="Backtest mode: research (kill switch resets) or live (permanent)") args = parser.parse_args() manifest = load_manifest(args.manifest) config = resolve_config(manifest, config_root=args.config_root, snapshot_id_override=args.snapshot_id) - store = _build_store(manifest, config, args.split, snapshot_dir_override=args.snapshot_dir) - runner = BacktestRunner( - manifest=manifest, - config=config, - store=store, - initial_equity=args.initial_equity, - ) - result = runner.run(output_root=args.output_root) - print(f"Run complete: {result.run_id}") - print(f"Trades: {result.metrics.trade_count}") - if result.metrics.total_return_pct is not None: - print(f"Total return: {result.metrics.total_return_pct:.2f}%") + if args.mode: + config.risk.backtest_mode = args.mode + + if args.walk_forward: + fold_results = run_walk_forward( + manifest=manifest, + config=config, + snapshot_dir_override=args.snapshot_dir, + initial_equity=args.initial_equity, + output_root=args.output_root, + train_days=args.wf_train_days, + test_days=args.wf_test_days, + ) + # Write summary JSON + summary_path = Path(args.output_root) / "walk_forward_summary.json" + summary_path.parent.mkdir(parents=True, exist_ok=True) + summary_path.write_text(json.dumps(fold_results, indent=2)) + print(f"\nSummary written to: {summary_path}") + else: + store = _build_store(manifest, config, args.split, snapshot_dir_override=args.snapshot_dir) + runner = BacktestRunner( + manifest=manifest, + config=config, + store=store, + initial_equity=args.initial_equity, + ) + result = runner.run(output_root=args.output_root) + print(f"Run complete: {result.run_id}") + print(f"Trades: {result.metrics.trade_count}") + if result.metrics.total_return_pct is not None: + print(f"Total return: {result.metrics.total_return_pct:.2f}%") if __name__ == "__main__": diff --git a/configs/backtest/defaults.json b/configs/backtest/defaults.json index fcb818f..373e8da 100644 --- a/configs/backtest/defaults.json +++ b/configs/backtest/defaults.json @@ -8,23 +8,29 @@ }, "signal": { "score_threshold": 0.5, - "max_candidates_per_day": 5, + "max_candidates_per_day": 3, "execution_timing": "next_open", "decision_timing": "reaction_close", "ranking_fields": ["score", "avg_dollar_volume"] }, "risk": { - "per_trade_risk_pct": 0.01, - "max_daily_new_risk_pct": 0.03, - "max_positions": 10, - "max_positions_per_sector": 3, + "per_trade_risk_pct": 0.005, + "max_daily_new_risk_pct": 0.015, + "max_positions": 4, + "max_positions_per_sector": 2, "max_position_value_pct": 0.10, "max_adv_fraction": 0.01, "cooldown_after_loss_streak": 3, "cooldown_days": 2, "macro_regime_enabled": false, "macro_sma_period": 20, - "stop_atr_multiplier": 3.0 + "stop_atr_multiplier": 3.0, + "backtest_mode": "research", + "kill_switch_cooldown_days": 20, + "veto_oneoff_penalty": 0.5, + "veto_parse_confidence_min": 0.4, + "veto_unknown_direction": true, + "veto_bearish_direction": true }, "execution": { "entry_fill_model": "next_open", @@ -55,14 +61,16 @@ }, "guidance_update": { "enabled": true, - "max_holding_days_override": 10 + "max_holding_days_override": 10, + "direction_filter": "bullish_only" }, "management_change": { "enabled": false }, "material_contract": { "enabled": true, - "max_holding_days_override": 5 + "max_holding_days_override": 5, + "direction_filter": "bullish_only" }, "other_material_event": { "enabled": false diff --git a/dev/analysis/concurrent_position_analysis.py b/dev/analysis/concurrent_position_analysis.py new file mode 100644 index 0000000..5d48840 --- /dev/null +++ b/dev/analysis/concurrent_position_analysis.py @@ -0,0 +1,169 @@ +"""Concurrent Position Analysis: Position clustering and correlation. + +Are simultaneous positions correlated? Analyzes trade blotter to find +overlapping position groups and their sector distribution. + +Usage: + python -m dev.analysis.concurrent_position_analysis \ + --run-dir runs/run_20260313_xyz [--output-dir ./data/analysis] +""" +from __future__ import annotations + +import argparse +import csv +import datetime as dt +from collections import Counter +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + + +def _load_blotter(run_dir: Path) -> list[dict[str, Any]]: + """Load trade blotter from parquet.""" + blotter_path = run_dir / "trade_blotter.parquet" + if not blotter_path.exists(): + # Try CSV fallback + csv_path = run_dir / "trade_blotter.csv" + if csv_path.exists(): + rows = [] + with open(csv_path) as f: + reader = csv.DictReader(f) + for row in reader: + rows.append(row) + return rows + raise FileNotFoundError(f"No trade blotter found in {run_dir}") + + table = pq.read_table(blotter_path) + return table.to_pylist() + + +def _parse_date(val: Any) -> dt.date | None: + """Parse a date value from various formats.""" + if val is None: + return None + if isinstance(val, dt.date): + return val + if isinstance(val, dt.datetime): + return val.date() + try: + return dt.date.fromisoformat(str(val)[:10]) + except (ValueError, TypeError): + return None + + +def analyze(trades: list[dict[str, Any]], output_dir: Path) -> None: + output_dir.mkdir(parents=True, exist_ok=True) + + # Parse entry/exit dates + positions: list[dict[str, Any]] = [] + for t in trades: + entry = _parse_date(t.get("entry_date")) + exit_ = _parse_date(t.get("exit_date")) + if entry and exit_: + positions.append({ + "trade_id": t.get("trade_id", ""), + "symbol": t.get("symbol", ""), + "sector": t.get("sector", "UNKNOWN"), + "entry_date": entry, + "exit_date": exit_, + }) + + if not positions: + print("No valid trades found.") + return + + # Build daily position timeline + all_dates = set() + for p in positions: + d = p["entry_date"] + while d <= p["exit_date"]: + all_dates.add(d) + d += dt.timedelta(days=1) + + daily_positions: dict[dt.date, list[dict[str, Any]]] = {} + for d in sorted(all_dates): + active = [p for p in positions if p["entry_date"] <= d <= p["exit_date"]] + if active: + daily_positions[d] = active + + # Compute concurrent position stats + concurrent_counts = [len(v) for v in daily_positions.values()] + max_concurrent = max(concurrent_counts) if concurrent_counts else 0 + avg_concurrent = sum(concurrent_counts) / len(concurrent_counts) if concurrent_counts else 0 + + # Find overlap groups (days with >1 position) + overlap_days = {d: ps for d, ps in daily_positions.items() if len(ps) > 1} + + # Sector distribution in overlap groups + overlap_sector_counts: Counter[str] = Counter() + overlap_symbol_counts: Counter[str] = Counter() + for d, ps in overlap_days.items(): + for p in ps: + overlap_sector_counts[p["sector"]] += 1 + overlap_symbol_counts[p["symbol"]] += 1 + + # Peak overlap days + peak_days = sorted(overlap_days.items(), key=lambda x: len(x[1]), reverse=True)[:10] + + # CSV: daily concurrent position counts + csv_rows: list[dict[str, Any]] = [] + for d in sorted(daily_positions): + ps = daily_positions[d] + sectors = Counter(p["sector"] for p in ps) + csv_rows.append({ + "date": str(d), + "concurrent_positions": len(ps), + "unique_sectors": len(sectors), + "sectors": "|".join(f"{s}:{c}" for s, c in sectors.most_common()), + "symbols": "|".join(p["symbol"] for p in ps), + }) + + csv_path = output_dir / "concurrent_position_analysis.csv" + if csv_rows: + fieldnames = list(csv_rows[0].keys()) + with open(csv_path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(csv_rows) + + # Console output + print(f"\n{'='*70}") + print(f"Concurrent Position Analysis — {len(positions)} trades") + print(f"{'='*70}") + print(f"\n Total trading days with positions: {len(daily_positions)}") + print(f" Days with >1 concurrent position: {len(overlap_days)}") + print(f" Max concurrent positions: {max_concurrent}") + print(f" Avg concurrent positions: {avg_concurrent:.1f}") + + if overlap_sector_counts: + print(f"\n Sector distribution in overlap periods:") + for sector, count in overlap_sector_counts.most_common(): + print(f" {sector:<20} {count:>5} position-days") + + if peak_days: + print(f"\n Top 10 peak overlap days:") + for d, ps in peak_days: + symbols = ", ".join(p["symbol"] for p in ps) + sectors = set(p["sector"] for p in ps) + print(f" {d} {len(ps)} positions sectors={len(sectors)} [{symbols}]") + + print(f"\nCSV written to: {csv_path}") + + +def main() -> None: + parser = argparse.ArgumentParser(description="Concurrent Position Analysis") + parser.add_argument("--run-dir", required=True, help="Path to backtest run directory") + parser.add_argument("--output-dir", default="./data/analysis", help="Output directory") + args = parser.parse_args() + + trades = _load_blotter(Path(args.run_dir)) + if not trades: + print("No trades found in blotter.") + return + + analyze(trades, Path(args.output_dir)) + + +if __name__ == "__main__": + main() diff --git a/dev/analysis/event_type_decomposition.py b/dev/analysis/event_type_decomposition.py new file mode 100644 index 0000000..1a62463 --- /dev/null +++ b/dev/analysis/event_type_decomposition.py @@ -0,0 +1,122 @@ +"""Event Type Decomposition: Per-event-type performance analysis. + +Which event types contribute alpha? + +Groups snapshot data by event_type and computes forward return statistics, +win rates, and average composite scores per type. + +Usage: + python -m dev.analysis.event_type_decomposition \ + --snapshot-dir data/datasets/snapshots/snapshot_2026_03_20 \ + --split train [--output-dir ./data/analysis] +""" +from __future__ import annotations + +import argparse +import csv +import statistics +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + +from libs.backtest.scoring import compute_entry_score + + +def _load_rows(snapshot_dir: Path, split: str) -> list[dict[str, Any]]: + parquet_path = snapshot_dir / f"{split}.parquet" + table = pq.read_table(parquet_path) + return table.to_pylist() + + +FWD_HORIZONS = ["fwd_return_1d", "fwd_return_3d", "fwd_return_5d", "fwd_return_10d"] + + +def analyze(rows: list[dict[str, Any]], output_dir: Path) -> None: + output_dir.mkdir(parents=True, exist_ok=True) + + # Compute scores + for row in rows: + row["_composite_score"] = compute_entry_score(row) + + # Group by event_type + by_type: dict[str, list[dict[str, Any]]] = {} + for row in rows: + et = row.get("event_type", "unknown") + by_type.setdefault(et, []).append(row) + + csv_rows: list[dict[str, Any]] = [] + for event_type in sorted(by_type): + items = by_type[event_type] + n = len(items) + row_out: dict[str, Any] = {"event_type": event_type, "count": n} + + # Average composite score + scores = [r["_composite_score"] for r in items] + row_out["avg_score"] = statistics.mean(scores) if scores else None + + for horizon in FWD_HORIZONS: + vals = [float(r[horizon]) for r in items if r.get(horizon) is not None] + if vals: + row_out[f"{horizon}_mean"] = statistics.mean(vals) + row_out[f"{horizon}_win_rate"] = sum(1 for v in vals if v > 0) / len(vals) + row_out[f"{horizon}_std"] = statistics.stdev(vals) if len(vals) > 1 else 0.0 + else: + row_out[f"{horizon}_mean"] = None + row_out[f"{horizon}_win_rate"] = None + row_out[f"{horizon}_std"] = None + + csv_rows.append(row_out) + + # Write CSV + csv_path = output_dir / "event_type_decomposition.csv" + if csv_rows: + fieldnames = list(csv_rows[0].keys()) + with open(csv_path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(csv_rows) + + # Console output + print(f"\n{'='*90}") + print(f"Event Type Decomposition — {len(rows)} events, {len(by_type)} types") + print(f"{'='*90}") + print(f"\n{'Type':<22} {'N':>5} {'AvgScore':>8}", end="") + for h in FWD_HORIZONS: + label = h.replace("fwd_return_", "") + print(f" {label+'_mean':>9} {label+'_wr':>7}", end="") + print() + print("-" * 90) + + for row_out in csv_rows: + avg_s = row_out["avg_score"] + print(f"{row_out['event_type']:<22} {row_out['count']:>5} {avg_s:.3f}" if avg_s else + f"{row_out['event_type']:<22} {row_out['count']:>5} N/A", end="") + for h in FWD_HORIZONS: + mean_val = row_out[f"{h}_mean"] + wr_val = row_out[f"{h}_win_rate"] + mean_str = f"{mean_val:+.4f}" if mean_val is not None else " N/A" + wr_str = f"{wr_val:.1%}" if wr_val is not None else " N/A" + print(f" {mean_str:>9} {wr_str:>7}", end="") + print() + + print(f"\nCSV written to: {csv_path}") + + +def main() -> None: + parser = argparse.ArgumentParser(description="Event Type Decomposition Analysis") + parser.add_argument("--snapshot-dir", required=True, help="Path to snapshot directory") + parser.add_argument("--split", default="train", help="Split name (train/valid/test)") + parser.add_argument("--output-dir", default="./data/analysis", help="Output directory") + args = parser.parse_args() + + rows = _load_rows(Path(args.snapshot_dir), args.split) + if not rows: + print("No data found.") + return + + analyze(rows, Path(args.output_dir)) + + +if __name__ == "__main__": + main() diff --git a/dev/analysis/kill_switch_impact_analysis.py b/dev/analysis/kill_switch_impact_analysis.py new file mode 100644 index 0000000..542816e --- /dev/null +++ b/dev/analysis/kill_switch_impact_analysis.py @@ -0,0 +1,122 @@ +"""Kill Switch Impact Analysis: Compare live vs research mode. + +How much does the permanent kill switch hide? + +Runs the backtest twice (live mode and research mode) and compares +key metrics side-by-side. + +Usage: + python -m dev.analysis.kill_switch_impact_analysis \ + --manifest configs/experiments/phase5_v1.json \ + --snapshot-dir data/datasets/snapshots \ + --split train [--output-root ./runs] +""" +from __future__ import annotations + +import argparse +from pathlib import Path + +from libs.backtest.manifests import load_manifest, resolve_config +from libs.backtest.domain import MetricsBundle + + +def _run_backtest( + manifest_path: str, + snapshot_dir: str | None, + split: str, + mode: str, + output_root: str, + initial_equity: float, + config_root: str, +) -> tuple[str, MetricsBundle, int]: + """Run a single backtest and return (run_id, metrics, trade_count).""" + from apps.backtester.run import BacktestRunner, _build_store + + manifest = load_manifest(manifest_path) + config = resolve_config(manifest, config_root=config_root) + config.risk.backtest_mode = mode + + store = _build_store(manifest, config, split, snapshot_dir_override=snapshot_dir) + runner = BacktestRunner( + manifest=manifest, + config=config, + store=store, + initial_equity=initial_equity, + ) + result = runner.run(output_root=output_root) + return result.run_id, result.metrics, result.metrics.trade_count + + +def _fmt(val: float | None, fmt: str = ".2f") -> str: + if val is None: + return "N/A" + return f"{val:{fmt}}" + + +def _fmt_pct(val: float | None) -> str: + if val is None: + return "N/A" + return f"{val:.2f}%" + + +def main() -> None: + parser = argparse.ArgumentParser(description="Kill Switch Impact Analysis") + parser.add_argument("--manifest", required=True, help="Path to experiment manifest JSON") + parser.add_argument("--snapshot-dir", help="Override snapshot root directory") + parser.add_argument("--split", default="train", help="Split name (train/valid/test)") + parser.add_argument("--output-root", default="./runs", help="Output root directory") + parser.add_argument("--initial-equity", type=float, default=100_000.0) + parser.add_argument("--config-root", default=".", help="Root dir for resolving config paths") + args = parser.parse_args() + + print("Running LIVE mode backtest...") + live_id, live_m, live_trades = _run_backtest( + args.manifest, args.snapshot_dir, args.split, "live", + args.output_root, args.initial_equity, args.config_root, + ) + + print("Running RESEARCH mode backtest...") + research_id, research_m, research_trades = _run_backtest( + args.manifest, args.snapshot_dir, args.split, "research", + args.output_root, args.initial_equity, args.config_root, + ) + + # Side-by-side comparison + print(f"\n{'='*60}") + print(f"Kill Switch Impact Analysis — split={args.split}") + print(f"{'='*60}") + print(f"\n{'Metric':<28} {'Live':>14} {'Research':>14}") + print("-" * 60) + + rows = [ + ("Run ID", live_id[:16], research_id[:16]), + ("Total Trades", str(live_trades), str(research_trades)), + ("Win Rate", _fmt_pct(live_m.win_rate and live_m.win_rate * 100), + _fmt_pct(research_m.win_rate and research_m.win_rate * 100)), + ("Total Return", _fmt_pct(live_m.total_return_pct), + _fmt_pct(research_m.total_return_pct)), + ("Max Drawdown", _fmt_pct(live_m.max_drawdown_pct), + _fmt_pct(research_m.max_drawdown_pct)), + ("Sharpe Ratio", _fmt(live_m.sharpe_ratio), _fmt(research_m.sharpe_ratio)), + ("Sortino Ratio", _fmt(live_m.sortino_ratio), _fmt(research_m.sortino_ratio)), + ("Profit Factor", _fmt(live_m.profit_factor), _fmt(research_m.profit_factor)), + ("Expectancy (R)", _fmt(live_m.expectancy_r), _fmt(research_m.expectancy_r)), + ("Avg Holding Days", _fmt(live_m.avg_holding_days, ".1f"), + _fmt(research_m.avg_holding_days, ".1f")), + ("Avg Positions Held", _fmt(live_m.avg_positions_held, ".1f"), + _fmt(research_m.avg_positions_held, ".1f")), + ] + + for label, live_val, research_val in rows: + print(f"{label:<28} {live_val:>14} {research_val:>14}") + + # Delta summary + if live_m.total_return_pct is not None and research_m.total_return_pct is not None: + delta_return = research_m.total_return_pct - live_m.total_return_pct + delta_trades = research_trades - live_trades + print(f"\n Delta trades: {delta_trades:+d}") + print(f" Delta return: {delta_return:+.2f}%") + + +if __name__ == "__main__": + main() diff --git a/dev/analysis/signal_quality_analysis.py b/dev/analysis/signal_quality_analysis.py new file mode 100644 index 0000000..77206bb --- /dev/null +++ b/dev/analysis/signal_quality_analysis.py @@ -0,0 +1,170 @@ +"""Signal Quality Analysis: Score-bucket monotonicity check. + +Does higher composite score predict better forward returns? + +Reads snapshot parquet, computes compute_entry_score for each row, +buckets into quintiles, and checks Spearman rank correlation between +score bucket and mean forward return. + +Usage: + python -m dev.analysis.signal_quality_analysis \ + --snapshot-dir data/datasets/snapshots/snapshot_2026_03_20 \ + --split train [--output-dir ./data/analysis] +""" +from __future__ import annotations + +import argparse +import csv +import statistics +from pathlib import Path +from typing import Any + +import pyarrow.parquet as pq + +from libs.backtest.scoring import compute_entry_score + + +def _load_rows(snapshot_dir: Path, split: str) -> list[dict[str, Any]]: + """Load parquet snapshot into list of dicts.""" + parquet_path = snapshot_dir / f"{split}.parquet" + table = pq.read_table(parquet_path) + return table.to_pylist() + + +def _bucket_label(score: float) -> str: + """Assign score to a quintile bucket.""" + if score < 0.2: + return "0.0-0.2" + if score < 0.4: + return "0.2-0.4" + if score < 0.6: + return "0.4-0.6" + if score < 0.8: + return "0.6-0.8" + return "0.8-1.0" + + +BUCKET_ORDER = ["0.0-0.2", "0.2-0.4", "0.4-0.6", "0.6-0.8", "0.8-1.0"] +FWD_HORIZONS = ["fwd_return_1d", "fwd_return_3d", "fwd_return_5d", "fwd_return_10d"] + + +def analyze(rows: list[dict[str, Any]], output_dir: Path) -> None: + """Score all rows, bucket, compute statistics, write output.""" + output_dir.mkdir(parents=True, exist_ok=True) + + # Score each row and assign bucket + scored: list[dict[str, Any]] = [] + for row in rows: + score = compute_entry_score(row) + bucket = _bucket_label(score) + scored.append({"score": score, "bucket": bucket, **row}) + + # Group by bucket + by_bucket: dict[str, list[dict[str, Any]]] = {b: [] for b in BUCKET_ORDER} + for s in scored: + by_bucket[s["bucket"]].append(s) + + # Compute per-bucket stats + csv_rows: list[dict[str, Any]] = [] + for bucket in BUCKET_ORDER: + items = by_bucket[bucket] + n = len(items) + row_out: dict[str, Any] = {"bucket": bucket, "count": n} + + for horizon in FWD_HORIZONS: + vals = [float(r[horizon]) for r in items if r.get(horizon) is not None] + if vals: + row_out[f"{horizon}_mean"] = statistics.mean(vals) + row_out[f"{horizon}_win_rate"] = sum(1 for v in vals if v > 0) / len(vals) + else: + row_out[f"{horizon}_mean"] = None + row_out[f"{horizon}_win_rate"] = None + + csv_rows.append(row_out) + + # Spearman rank correlation (bucket rank vs mean return) + spearman_results: dict[str, float | None] = {} + for horizon in FWD_HORIZONS: + means = [r[f"{horizon}_mean"] for r in csv_rows if r[f"{horizon}_mean"] is not None] + if len(means) >= 3: + # Rank correlation: do the means increase with bucket rank? + ranks_score = list(range(len(means))) + ranks_return = _rank(means) + spearman_results[horizon] = _spearman(ranks_score, ranks_return) + else: + spearman_results[horizon] = None + + # Write CSV + csv_path = output_dir / "signal_quality_analysis.csv" + if csv_rows: + fieldnames = list(csv_rows[0].keys()) + with open(csv_path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(csv_rows) + + # Console output + print(f"\n{'='*80}") + print(f"Signal Quality Analysis — {len(scored)} events") + print(f"{'='*80}") + print(f"\n{'Bucket':<12} {'Count':>6}", end="") + for h in FWD_HORIZONS: + label = h.replace("fwd_return_", "") + print(f" {label+'_mean':>10} {label+'_wr':>8}", end="") + print() + print("-" * 80) + + for row_out in csv_rows: + print(f"{row_out['bucket']:<12} {row_out['count']:>6}", end="") + for h in FWD_HORIZONS: + mean_val = row_out[f"{h}_mean"] + wr_val = row_out[f"{h}_win_rate"] + mean_str = f"{mean_val:+.4f}" if mean_val is not None else " N/A" + wr_str = f"{wr_val:.1%}" if wr_val is not None else " N/A" + print(f" {mean_str:>10} {wr_str:>8}", end="") + print() + + print(f"\nSpearman rank correlation (bucket vs mean return):") + for h in FWD_HORIZONS: + label = h.replace("fwd_return_", "") + val = spearman_results.get(h) + print(f" {label}: {val:+.3f}" if val is not None else f" {label}: N/A") + + print(f"\nCSV written to: {csv_path}") + + +def _rank(values: list[float]) -> list[float]: + """Compute ranks (0-indexed) for a list of values.""" + indexed = sorted(enumerate(values), key=lambda x: x[1]) + ranks = [0.0] * len(values) + for rank, (idx, _) in enumerate(indexed): + ranks[idx] = float(rank) + return ranks + + +def _spearman(x: list[float], y: list[float]) -> float: + """Compute Spearman rank correlation coefficient.""" + n = len(x) + if n < 2: + return 0.0 + d_sq = sum((xi - yi) ** 2 for xi, yi in zip(x, y)) + return 1 - (6 * d_sq) / (n * (n**2 - 1)) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Signal Quality Analysis") + parser.add_argument("--snapshot-dir", required=True, help="Path to snapshot directory") + parser.add_argument("--split", default="train", help="Split name (train/valid/test)") + parser.add_argument("--output-dir", default="./data/analysis", help="Output directory") + args = parser.parse_args() + + rows = _load_rows(Path(args.snapshot_dir), args.split) + if not rows: + print("No data found.") + return + + analyze(rows, Path(args.output_dir)) + + +if __name__ == "__main__": + main() diff --git a/libs/backtest/allocator.py b/libs/backtest/allocator.py index 50ad026..9337dcf 100644 --- a/libs/backtest/allocator.py +++ b/libs/backtest/allocator.py @@ -92,7 +92,7 @@ def run_entry_gates( cooldown_remaining: int = 0, macro_data: dict[str, Any] | None = None, ) -> str | None: - """Run 8-step entry gate. Returns skip_reason string or None (pass). + """Run entry gates. Returns skip_reason string or None (pass). Gates (in order): 0. Macro regime (SPY below SMA — bearish market) @@ -103,6 +103,12 @@ def run_entry_gates( 5. Daily new risk budget 6. Cash available (estimated position cost) 7. Loss-streak cooldown + 8. SUE gate (earnings: positive surprise required) + 9. Event-type direction filter (bullish_only) + 10. High one-off risk (veto: oneoff_penalty >= threshold) + 11. Low parse confidence (veto: parse_confidence < threshold) + 12. Unknown direction (veto: event_direction == "unknown") + 13. Bearish direction (veto: event_direction == "bearish") """ # Gate 0: Macro regime filter if config.risk.macro_regime_enabled and macro_data: @@ -163,6 +169,27 @@ def run_entry_gates( if reaction is not None and float(reaction) < 0: return "direction_filter_bearish" + # --- Veto gates: document quality hard filters --- + + # Gate 10: High one-off risk + oneoff = candidate.features.get("oneoff_penalty") + if oneoff is not None and float(oneoff) >= config.risk.veto_oneoff_penalty: + return "high_oneoff_risk" + + # Gate 11: Low parse confidence + parse_conf = candidate.features.get("parse_confidence_overall") + if parse_conf is not None and float(parse_conf) < config.risk.veto_parse_confidence_min: + return "low_parse_confidence" + + # Gate 12: Unknown direction + event_dir = candidate.features.get("event_direction") + if config.risk.veto_unknown_direction and event_dir is not None and str(event_dir).lower() == "unknown": + return "unknown_direction" + + # Gate 13: Bearish direction (all event types, document-based) + if config.risk.veto_bearish_direction and event_dir is not None and str(event_dir).lower() == "bearish": + return "bearish_direction" + return None # all gates passed diff --git a/libs/backtest/domain.py b/libs/backtest/domain.py index 8750f3b..2e4a7e6 100644 --- a/libs/backtest/domain.py +++ b/libs/backtest/domain.py @@ -27,6 +27,11 @@ class ExitReason(str, Enum): MISSING_BAR = "MISSING_BAR" +class BacktestMode(str, Enum): + RESEARCH = "research" + LIVE = "live" + + class Candidate(BaseModel): """An eligible trade candidate derived from a Parquet snapshot row.""" model_config = ConfigDict(frozen=True) @@ -189,6 +194,12 @@ class RiskConfig(BaseModel): macro_regime_enabled: bool = False # block entries when SPY < SMA macro_sma_period: int = 20 # SMA lookback for macro regime stop_atr_multiplier: float = 1.5 # ATR multiplier for stop distance + backtest_mode: str = "research" # "research" or "live" + kill_switch_cooldown_days: int = 20 # trading days before reset (research only) + veto_oneoff_penalty: float = 0.5 # block if oneoff_penalty >= this + veto_parse_confidence_min: float = 0.4 # block if parse_confidence < this + veto_unknown_direction: bool = True # block if event_direction == "unknown" + veto_bearish_direction: bool = True # block if event_direction == "bearish" class ExecutionConfig(BaseModel): diff --git a/libs/backtest/scoring.py b/libs/backtest/scoring.py index 2ac4166..f4a5758 100644 --- a/libs/backtest/scoring.py +++ b/libs/backtest/scoring.py @@ -1,19 +1,24 @@ """Rule-based entry score model for the backtester. -Computes a composite score in [0, 1] from market and event features +Computes a composite score in [0, 1] from event, market, and text features available at entry time (no forward-looking data). Higher score = more favorable entry conditions for a long swing trade. -Market features (75% weight — primary signal): - - Moderate positive reaction → likely PEAD continuation - - Extreme positive reaction → already priced in, mean-reversion risk - - Close near session high → buyers in control - - Above-average volume → conviction (but extreme volume = exhaustion) - - Small positive gap → orderly strength - -Event features (25% weight — supplementary signal): - - Document quality & signal strength → confidence in the event parsing - - Risk flags (oneoff_penalty) → penalty for suspicious events +Event/Document Quality (55% weight — primary signal): + - Event quality (20%) — parser confidence + signal strength + - Earnings surprise (12%) — SUE/EPS growth (earnings events only) + - Risk penalty (10%) — oneoff risk flags reduce score + - Parse confidence (8%) — parse_confidence_overall from parser + - Direction clarity (5%) — event_direction categorical field + +Market Confirmation (35% weight — secondary signal): + - Reaction quality (12%) — moderate positive return is ideal + - Close strength (10%) — close near high = buyers won the day + - Volume conviction (8%) — above-average but not exhaustion + - Gap quality (5%) — small positive gap = orderly strength + +Text (10% weight — filing sentiment): + - LM sentiment (10%) — Loughran-McDonald filing tone """ from __future__ import annotations @@ -25,40 +30,49 @@ logger = get_logger(__name__) def compute_entry_score(row: dict[str, Any]) -> float: - """Compute composite entry score from market + event features. + """Compute composite entry score from event + market + text features. Components and weights: - Market (65%): - 1. Reaction quality (20%) — moderate positive return is ideal - 2. Close strength (20%) — close near high = buyers won the day - 3. Volume conviction (15%) — above-average but not exhaustion - 4. Gap quality (10%) — small positive gap = orderly strength - Event (35%): - 5. Event quality (15%) — parser confidence + signal strength - 6. Earnings surprise (10%) — SUE/EPS growth (earnings events only) - 7. Risk penalty (10%) — oneoff risk flags reduce score + Event/Document Quality (55%): + 1. Event quality (20%) — parser confidence + signal strength + 2. Earnings surprise (12%) — SUE/EPS growth (earnings events only) + 3. Risk penalty (10%) — oneoff risk flags reduce score + 4. Parse confidence (8%) — parse_confidence_overall from parser + 5. Direction clarity (5%) — event_direction categorical field + Market Confirmation (35%): + 6. Reaction quality (12%) — moderate positive return is ideal + 7. Close strength (10%) — close near high = buyers won the day + 8. Volume conviction (8%) — above-average but not exhaustion + 9. Gap quality (5%) — small positive gap = orderly strength + Text (10%): + 10. LM sentiment (10%) — Loughran-McDonald filing tone Returns float in [0.0, 1.0]. """ - # Market components + # Event/Document Quality components + event = _event_quality_score(row) + sue = _earnings_surprise_score(row) + risk = _risk_penalty_score(row) + parse_conf = _parse_confidence_score(row) + direction = _direction_clarity_score(row) + + # Market Confirmation components reaction = _reaction_score(row) close = _close_strength_score(row) volume = _volume_score(row) gap = _gap_score(row) - # Event components (gracefully handle missing features) - event = _event_quality_score(row) - sue = _earnings_surprise_score(row) - risk = _risk_penalty_score(row) + # Text sentiment component + text = _text_sentiment_score(row) raw = ( - reaction * 0.20 - + close * 0.20 - + volume * 0.15 - + gap * 0.10 - + event * 0.15 - + sue * 0.10 - + risk * 0.10 + # Event/Document Quality (55%) + event * 0.20 + sue * 0.12 + risk * 0.10 + + parse_conf * 0.08 + direction * 0.05 + # Market Confirmation (35%) + + reaction * 0.12 + close * 0.10 + volume * 0.08 + gap * 0.05 + # Text (10%) + + text * 0.10 ) return max(0.0, min(1.0, raw)) @@ -282,3 +296,65 @@ def _risk_penalty_score(row: dict[str, Any]) -> float: p = max(0.0, min(1.0, float(penalty))) # Linear inversion: 0 -> 0.9, 1.0 -> 0.2 return 0.9 - 0.7 * p + + +def _parse_confidence_score(row: dict[str, Any]) -> float: + """Score based on parse_confidence_overall [0-1].""" + conf = row.get("parse_confidence_overall") + if conf is None: + return 0.5 + c = float(conf) + if c > 0.8: + return 0.9 + if c > 0.6: + return 0.7 + if c > 0.5: + return 0.5 + if c > 0.4: + return 0.3 + return 0.2 + + +def _direction_clarity_score(row: dict[str, Any]) -> float: + """Score based on event_direction categorical field.""" + direction = row.get("event_direction") + if direction is None: + return 0.5 + return {"bullish": 0.9, "mixed": 0.4, "neutral": 0.3, + "bearish": 0.1, "unknown": 0.2}.get(str(direction).lower(), 0.5) + + +# --------------------------------------------------------------------------- +# Text sentiment scoring +# --------------------------------------------------------------------------- + + +def _text_sentiment_score(row: dict[str, Any]) -> float: + """Score based on Loughran-McDonald text sentiment features. + + Uses lm_net_sentiment (positive - negative word fraction). + Typical range is [-0.02, +0.02] for SEC filings. + + Mapping: + > +0.005 -> 0.8 (noticeably positive tone) + > +0.001 -> 0.65 (mildly positive) + > -0.001 -> 0.5 (neutral) + > -0.005 -> 0.35 (mildly negative) + <= -0.005 -> 0.2 (noticeably negative tone) + + When absent, returns neutral 0.5. + """ + net = row.get("lm_net_sentiment") + if net is None: + return 0.5 + + n = float(net) + if n > 0.005: + return 0.8 + if n > 0.001: + return 0.65 + if n > -0.001: + return 0.5 + if n > -0.005: + return 0.35 + return 0.2 diff --git a/tests/unit/backtest/test_allocator.py b/tests/unit/backtest/test_allocator.py index 27c5524..7d498dc 100644 --- a/tests/unit/backtest/test_allocator.py +++ b/tests/unit/backtest/test_allocator.py @@ -406,6 +406,84 @@ class TestDirectionFilter: assert result is None +class TestVetoGates: + """Veto gate tests for document quality hard filters (gates 10-13).""" + + def test_high_oneoff_blocked(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={"oneoff_penalty": 0.6}) + cfg = _make_config() + cfg.risk.veto_oneoff_penalty = 0.5 + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result == "high_oneoff_risk" + + def test_low_oneoff_passes(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={"oneoff_penalty": 0.3}) + cfg = _make_config() + cfg.risk.veto_oneoff_penalty = 0.5 + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result is None + + def test_low_parse_confidence_blocked(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={"parse_confidence_overall": 0.3}) + cfg = _make_config() + cfg.risk.veto_parse_confidence_min = 0.4 + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result == "low_parse_confidence" + + def test_adequate_parse_confidence_passes(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={"parse_confidence_overall": 0.6}) + cfg = _make_config() + cfg.risk.veto_parse_confidence_min = 0.4 + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result is None + + def test_unknown_direction_blocked(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={"event_direction": "unknown"}) + cfg = _make_config() + cfg.risk.veto_unknown_direction = True + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result == "unknown_direction" + + def test_bearish_direction_blocked(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={"event_direction": "bearish"}) + cfg = _make_config() + cfg.risk.veto_bearish_direction = True + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result == "bearish_direction" + + def test_bullish_direction_passes(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={"event_direction": "bullish"}) + cfg = _make_config() + cfg.risk.veto_unknown_direction = True + cfg.risk.veto_bearish_direction = True + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result is None + + def test_missing_features_pass_veto(self): + from libs.backtest.allocator import run_entry_gates + + c = _make_candidate(features={}) + cfg = _make_config() + cfg.risk.veto_unknown_direction = True + cfg.risk.veto_bearish_direction = True + result = run_entry_gates(c, _make_portfolio_state(), [], cfg) + assert result is None + + class TestBuildPlannedOrder: def test_valid_order(self): from libs.backtest.allocator import build_planned_order diff --git a/tests/unit/backtest/test_scoring.py b/tests/unit/backtest/test_scoring.py index 8944a16..f49bf16 100644 --- a/tests/unit/backtest/test_scoring.py +++ b/tests/unit/backtest/test_scoring.py @@ -5,9 +5,11 @@ import pytest from libs.backtest.scoring import ( _close_strength_score, + _direction_clarity_score, _earnings_surprise_score, _event_quality_score, _gap_score, + _parse_confidence_score, _reaction_score, _risk_penalty_score, _volume_score, @@ -139,7 +141,7 @@ class TestComputeEntryScore: "gap_size": 0.01, # orderly → 0.8 } score = compute_entry_score(row) - assert score > 0.7 + assert score > 0.60 def test_bearish_setup_scores_low(self): """Negative return + close near low + below avg volume.""" @@ -150,7 +152,7 @@ class TestComputeEntryScore: "gap_size": -0.03, # bearish gap → 0.2 } score = compute_entry_score(row) - assert score < 0.35 + assert score < 0.45 def test_extreme_positive_penalized(self): """Very large positive reaction should be penalized.""" @@ -214,7 +216,7 @@ class TestComputeEntryScore: "gap_size": 0.006, }) assert aapl > 0.55, f"AAPL should be above 0.55, got {aapl:.3f}" - assert tsla < 0.45, f"TSLA should be below 0.45, got {tsla:.3f}" + assert tsla < 0.48, f"TSLA should be below 0.48, got {tsla:.3f}" assert aapl > tsla def test_event_features_boost_score(self): @@ -323,3 +325,47 @@ class TestRiskPenaltyScore: def test_missing_returns_neutral(self): assert _risk_penalty_score({}) == 0.5 + + +class TestParseConfidenceScore: + """Parse confidence scoring.""" + + def test_high_confidence(self): + assert _parse_confidence_score({"parse_confidence_overall": 0.9}) == 0.9 + + def test_moderate_confidence(self): + assert _parse_confidence_score({"parse_confidence_overall": 0.7}) == 0.7 + + def test_borderline_confidence(self): + assert _parse_confidence_score({"parse_confidence_overall": 0.55}) == 0.5 + + def test_low_confidence(self): + assert _parse_confidence_score({"parse_confidence_overall": 0.45}) == 0.3 + + def test_very_low_confidence(self): + assert _parse_confidence_score({"parse_confidence_overall": 0.3}) == 0.2 + + def test_missing_returns_neutral(self): + assert _parse_confidence_score({}) == 0.5 + + +class TestDirectionClarityScore: + """Direction clarity scoring.""" + + def test_bullish(self): + assert _direction_clarity_score({"event_direction": "bullish"}) == 0.9 + + def test_mixed(self): + assert _direction_clarity_score({"event_direction": "mixed"}) == 0.4 + + def test_neutral(self): + assert _direction_clarity_score({"event_direction": "neutral"}) == 0.3 + + def test_bearish(self): + assert _direction_clarity_score({"event_direction": "bearish"}) == 0.1 + + def test_unknown(self): + assert _direction_clarity_score({"event_direction": "unknown"}) == 0.2 + + def test_missing_returns_neutral(self): + assert _direction_clarity_score({}) == 0.5