jackyu66gitandClaude
71951019fb
Phase A-C complete: 4 core factors, regime detection, signal tracking, Bayesian expectancy. chanmacro/ (32 files, ~4000 lines): - models: 12 enums + 15 Pydantic v2 models (DateAwareModel, MarketStateVector, etc.) - fetchers: OHLCV + Breadth (from data_provider) + Derivatives (new endpoint) - scoring: Price Structure / Breadth (quantile buckets) / OI Matrix (5 discrete states) / Volatility Regime - regime_detector: 3-state (TREND/RANGE/PANIC), factor-locked (Price+Breadth+Vol), versioned, 2-day confirmation - expectancy: SignalTracker (record+outcomes), TimeDecay (half-life=180d), BayesianExpectancyEngine (Empirical Bayes, Leveled, SufficiencyGuard) - validation: FactorValidator (IC/ICIR/Hit Ratio), RegimeValidator (MI/KL/ANOVA), TransitionValidator (stability) - CLI: fetch|score|regime|track|backfill|expectancy|validate|serve - tests: 52 passing (models, scoring, regime, expectancy) data_provider: - /api/derivatives endpoint: funding rate, OI, OI change, basis - _derivatives storage: same persist pattern as K-line (merge→lock→snapshot→atomic write) - background refresh every 60s Co-Authored-By: Claude <noreply@anthropic.com>
121 lines
4.7 KiB
Python
121 lines
4.7 KiB
Python
"""
|
|
validation/reporter.py — Aggregates all validation reports into a unified summary.
|
|
|
|
Used by: python main.py validate
|
|
"""
|
|
|
|
from datetime import date as Date
|
|
from typing import Optional
|
|
import logging
|
|
|
|
from .factor_validator import FactorValidator, FactorReport
|
|
from .regime_validator import RegimeValidator, RegimeReport
|
|
from .transition_validator import TransitionValidator, TransitionReport
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ValidationReporter:
|
|
"""
|
|
Orchestrates full validation pipeline:
|
|
|
|
1. Factor validation (IC, ICIR, Hit Ratio) for each factor
|
|
2. Regime validation (MI, KL, ANOVA) for each factor
|
|
3. Transition validation (stability, flip rate)
|
|
"""
|
|
|
|
def __init__(self, db_path: Optional[str] = None):
|
|
from config import config
|
|
self.db_path = db_path or config.db_path
|
|
self.factor_validator = FactorValidator(self.db_path)
|
|
self.regime_validator = RegimeValidator(self.db_path)
|
|
self.transition_validator = TransitionValidator(self.db_path)
|
|
|
|
def run_all(self) -> str:
|
|
"""Run all validations and return a formatted report string."""
|
|
lines = []
|
|
lines.append("=" * 70)
|
|
lines.append(f" ChanMacro Validation Report — {Date.today()}")
|
|
lines.append("=" * 70)
|
|
|
|
# ── Factor Validation ──────────────────────────
|
|
lines.append("")
|
|
lines.append("─" * 50)
|
|
lines.append(" FACTOR VALIDATION (Predictive Power)")
|
|
lines.append("─" * 50)
|
|
|
|
factor_queries = {
|
|
"Price Structure": "SELECT date, score FROM ohlcv_daily WHERE ema20 IS NOT NULL",
|
|
"Breadth": """
|
|
SELECT bd.date,
|
|
(bd.advance_top50*1.0/(bd.advance_top50+bd.decline_top50+1)*100*0.30
|
|
+ bd.above_ema20_top50*1.0/50*100*0.35
|
|
+ bd.new_highs_20d_top50*1.0/50*100*0.20
|
|
+ 50*0.15) as score
|
|
FROM breadth_daily bd
|
|
""",
|
|
}
|
|
|
|
factor_reports: list[FactorReport] = []
|
|
for name, query in factor_queries.items():
|
|
try:
|
|
report = self.factor_validator.validate_from_db(name, query)
|
|
factor_reports.append(report)
|
|
lines.append(report.summary())
|
|
lines.append("")
|
|
except Exception as e:
|
|
logger.warning(f"Factor validation failed for {name}: {e}")
|
|
|
|
# ── Regime Validation ──────────────────────────
|
|
lines.append("─" * 50)
|
|
lines.append(" REGIME VALIDATION (Regime Separation)")
|
|
lines.append("─" * 50)
|
|
|
|
regime_reports: list[RegimeReport] = []
|
|
for name, query in factor_queries.items():
|
|
try:
|
|
report = self.regime_validator.validate_from_db(name, query)
|
|
regime_reports.append(report)
|
|
lines.append(report.summary())
|
|
lines.append("")
|
|
except Exception as e:
|
|
logger.warning(f"Regime validation failed for {name}: {e}")
|
|
|
|
# ── Transition Validation ──────────────────────
|
|
lines.append("─" * 50)
|
|
lines.append(" TRANSITION VALIDATION (Regime Stability)")
|
|
lines.append("─" * 50)
|
|
|
|
try:
|
|
t_report = self.transition_validator.validate_from_db()
|
|
lines.append(t_report.summary())
|
|
except Exception as e:
|
|
logger.warning(f"Transition validation failed: {e}")
|
|
|
|
# ── Summary ────────────────────────────────────
|
|
lines.append("")
|
|
lines.append("=" * 70)
|
|
lines.append(" SUMMARY")
|
|
lines.append("=" * 70)
|
|
|
|
# Factor ranking by IC
|
|
if factor_reports:
|
|
ranked = sorted(factor_reports, key=lambda r: abs(r.ic_mean), reverse=True)
|
|
lines.append(" Factor Ranking (by |IC|):")
|
|
for i, r in enumerate(ranked):
|
|
tag = "★★★" if abs(r.ic_mean) > 0.05 else "★★" if abs(r.ic_mean) > 0.03 else "★"
|
|
lines.append(f" {i+1}. {r.factor_name:20s} IC={r.ic_mean:+.4f} {tag} {r.conclusion}")
|
|
|
|
# Regime factor ranking
|
|
if regime_reports:
|
|
ranked_r = sorted(regime_reports, key=lambda r: r.separation_score, reverse=True)
|
|
lines.append("")
|
|
lines.append(" Regime Factor Ranking (by Separation Score):")
|
|
for i, r in enumerate(ranked_r):
|
|
lines.append(f" {i+1}. {r.factor_name:20s} Score={r.separation_score:.2f} {r.conclusion}")
|
|
|
|
lines.append("")
|
|
lines.append("=" * 70)
|
|
|
|
return "\n".join(lines)
|