chanmacro: Signal Expectancy Engine V1 — Market Memory System

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>
This commit is contained in:
jackyu66git
2026-06-24 17:44:55 +08:00
co-authored by Claude
parent 34040575c1
commit 71951019fb
43 changed files with 5036 additions and 2 deletions
+130
View File
@@ -0,0 +1,130 @@
"""Test all Pydantic models and enums."""
import pytest
from datetime import date
from models import (
MarketRegime, OIState, BreadthBucket, VolRegime,
MarketStateVector, FactorScore, RegimeResult,
SignalFeatureRecord, ExpectancyReport, DailyOutput,
FactorContribution, SufficiencyLevel, SignalGrade,
)
class TestEnums:
def test_regime_values(self):
assert MarketRegime.TREND.value == "TREND"
assert MarketRegime.RANGE.value == "RANGE"
assert MarketRegime.PANIC.value == "PANIC"
def test_oi_state_has_neutral(self):
assert OIState.NEUTRAL.value == "Neutral"
assert len(OIState) == 5
def test_breadth_bucket_values(self):
assert BreadthBucket.EXTREME.value == "EXTREME"
assert len(BreadthBucket) == 5
def test_vol_regime_values(self):
assert VolRegime.LOW_VOL.value == "LOW_VOL"
assert VolRegime.EXPLOSIVE_VOL.value == "EXPLOSIVE_VOL"
class TestMarketStateVector:
def test_minimal_construction(self):
sv = MarketStateVector(
date="2026-06-24",
regime=MarketRegime.TREND,
regime_confidence=0.82,
regime_version="v1_price_breadth_vol",
)
assert sv.date == date(2026, 6, 24)
assert sv.regime == MarketRegime.TREND
assert sv.breadth_top50 == 50.0 # default
def test_date_string_parsing(self):
sv = MarketStateVector(
date="2026-01-15",
regime=MarketRegime.RANGE,
regime_confidence=0.55,
regime_version="v1_price_breadth_vol",
)
assert sv.date == date(2026, 1, 15)
def test_compute_hash(self):
sv = MarketStateVector(
date="2026-06-24",
regime=MarketRegime.TREND,
regime_confidence=0.82,
regime_version="v1_price_breadth_vol",
breadth_bucket=BreadthBucket.EXTREME,
oi_state=OIState.NEW_LONGS,
volatility_regime=VolRegime.NORMAL_VOL,
)
h = sv.compute_hash()
assert len(h) == 12
# Same state = same hash
sv2 = MarketStateVector(
date="2026-06-25",
regime=MarketRegime.TREND,
regime_confidence=0.80,
regime_version="v1_price_breadth_vol",
breadth_bucket=BreadthBucket.EXTREME,
oi_state=OIState.NEW_LONGS,
volatility_regime=VolRegime.NORMAL_VOL,
)
assert sv2.compute_hash() == h
def test_state_embedding(self):
sv = MarketStateVector(
date="2026-06-24",
regime=MarketRegime.TREND,
regime_confidence=0.82,
regime_version="v1_price_breadth_vol",
breadth_top20=80.0,
breadth_top30=75.0,
breadth_top50=70.0,
regime_maturity_score=60.0,
)
emb = sv.state_embedding()
assert len(emb) == 5
assert emb[0] == 80.0
assert emb[3] == 60.0
class TestRegimeResult:
def test_construction(self):
r = RegimeResult(
date="2026-06-24",
regime=MarketRegime.TREND,
confidence=0.82,
regime_version="v1_price_breadth_vol",
maturity_score=55.0,
all_scores={"TREND": 82.0, "RANGE": 45.0, "PANIC": 20.0},
confirmation_days=5,
)
assert r.regime == MarketRegime.TREND
assert r.confirmation_days == 5
class TestExpectancyReport:
def test_insufficient(self):
r = ExpectancyReport(
signal_type="B3",
date="2026-06-24",
final_estimate=0.0,
sufficiency=SufficiencyLevel.INSUFFICIENT,
source="insufficient",
)
assert r.final_estimate == 0.0
assert r.sufficiency == SufficiencyLevel.INSUFFICIENT
class TestFactorContribution:
def test_construction(self):
fc = FactorContribution(
factor="ETF Flow",
raw_score=85.0,
weight=0.1925,
impact=6.7,
direction="bullish",
)
assert fc.impact > 0