""" scoring/breadth_scorer.py — Market Breadth Score. The first citizen of the system. Diffusion always leads price. Multi-tier: Top20 / Top30 / Top50. Quantile-based bucketing: EXTREME / STRONG / NORMAL / WEAK / PANIC. 4 sub-indicators (equal weight): 1. Advance/Decline ratio (30%) 2. % above EMA20 (35%) 3. New 20d highs (20%) 4. BTC Dominance change (15%, inverted) """ from datetime import date as Date import sqlite3 import numpy as np import pandas as pd from .base import BaseScorer from .constants import ( BREADTH_W_ADVANCE, BREADTH_W_EMA20, BREADTH_W_NEW_HIGHS, BREADTH_W_BTC_DOM, ) from models import FactorScore, BreadthScore, BreadthBucket, MacroDirection from config import config class BreadthScorer(BaseScorer): """Scores market breadth with quantile-based bucketing.""" def compute(self, target_date: Date) -> BreadthScore: conn = self.get_connection() try: row = conn.execute( "SELECT * FROM breadth_daily WHERE date = ?", (str(target_date),) ).fetchone() if row is None: return BreadthScore( name="Breadth", score=50.0, label="No Data", breadth_bucket=BreadthBucket.NORMAL, ) row = dict(row) total = row.get("total_tracked", 50) or 50 # 1. Advance/Decline ratio advance = row.get("advance_top50", 0) or 0 decline = row.get("decline_top50", 0) or 0 if advance + decline > 0: ad_ratio = advance / (advance + decline) else: ad_ratio = 0.5 ad_score = ad_ratio * 100 # 2. % above EMA20 above_ema = row.get("above_ema20_top50", 0) or 0 ema_pct = above_ema / total if total > 0 else 0.5 ema_score = ema_pct * 100 # 3. New highs new_highs = row.get("new_highs_20d_top50", 0) or 0 highs_pct = new_highs / total if total > 0 else 0 highs_score = highs_pct * 100 # 4. BTC Dominance (inverted: BTC.D up = bearish for alts) btc_dom = row.get("btc_dominance") btc_dom_score = 50.0 # neutral default if btc_dom is not None: # Placeholder — needs historical comparison btc_dom_score = 50.0 # Weighted aggregate score = ( ad_score * BREADTH_W_ADVANCE + ema_score * BREADTH_W_EMA20 + highs_score * BREADTH_W_NEW_HIGHS + btc_dom_score * BREADTH_W_BTC_DOM ) # Multi-tier breadth b20 = self._compute_tier_breadth(row, 20, total) b30 = self._compute_tier_breadth(row, 30, total) b50 = score # Top50 = full score # Quantile bucket bucket = self._assign_bucket(score) # Divergence divergence = b20 - b50 # Direction if score >= 60: direction = MacroDirection.BULLISH elif score <= 40: direction = MacroDirection.BEARISH else: direction = MacroDirection.NEUTRAL # Narrative narrative = self._build_narrative(bucket, divergence, ema_pct, ad_ratio) return BreadthScore( name="Breadth", score=round(score, 1), label=bucket.value, direction=direction, breadth_top20=round(b20, 1), breadth_top30=round(b30, 1), breadth_top50=round(b50, 1), breadth_bucket=bucket, breadth_divergence=round(divergence, 1), advance_pct_top50=round(ad_ratio * 100, 1), above_ema20_pct_top50=round(ema_pct * 100, 1), new_highs_top50=new_highs, sub_scores={ "advance_decline": round(ad_score, 1), "above_ema20": round(ema_score, 1), "new_highs": round(highs_score, 1), "btc_dominance": round(btc_dom_score, 1), }, narrative=narrative, ) finally: conn.close() def _compute_tier_breadth(self, row: dict, tier: int, total: int) -> float: """Compute breadth score for a specific tier (Top20 or Top30).""" advance = row.get(f"advance_top{tier}", 0) or 0 above_ema = row.get(f"above_ema20_top{tier}", 0) or 0 new_highs = row.get(f"new_highs_20d_top{tier}", 0) or 0 tier_actual = min(tier, total) if tier_actual == 0: return 50.0 ad_ratio = advance / tier_actual if tier_actual > 0 else 0.5 ema_ratio = above_ema / tier_actual if tier_actual > 0 else 0.5 highs_ratio = new_highs / tier_actual if tier_actual > 0 else 0 return ( ad_ratio * 100 * BREADTH_W_ADVANCE + ema_ratio * 100 * BREADTH_W_EMA20 + highs_ratio * 100 * BREADTH_W_NEW_HIGHS + 50 * BREADTH_W_BTC_DOM # neutral for BTC.D ) def _assign_bucket(self, score: float) -> BreadthBucket: """Assign quantile-based bucket. V1 uses fixed thresholds until history accumulated.""" # V1: fixed thresholds (will switch to quantile when enough history) if score >= 80: return BreadthBucket.EXTREME elif score >= 60: return BreadthBucket.STRONG elif score >= 40: return BreadthBucket.NORMAL elif score >= 20: return BreadthBucket.WEAK else: return BreadthBucket.PANIC @staticmethod def compute_quantile_boundaries(db_path: str) -> dict: """Compute quantile boundaries from historical breadth data. This should be called after accumulating enough history (> 1 year). Returns boundaries for pd.qcut. """ conn = sqlite3.connect(db_path) df = pd.read_sql_query( "SELECT date, advance_top50, decline_top50, above_ema20_top50 FROM breadth_daily", conn ) conn.close() if len(df) < 100: return {"boundaries": [0, 20, 40, 60, 80, 100], "is_quantile": False} df["ad_ratio"] = df["advance_top50"] / (df["advance_top50"] + df["decline_top50"]) df["ema_ratio"] = df["above_ema20_top50"] / 50 df["breadth_raw"] = ( df["ad_ratio"] * BREADTH_W_ADVANCE * 100 + df["ema_ratio"] * BREADTH_W_EMA20 * 100 + 40 * BREADTH_W_NEW_HIGHS + 50 * BREADTH_W_BTC_DOM ) boundaries = list(np.percentile(df["breadth_raw"].dropna(), [10, 30, 70, 90])) return { "boundaries": [0] + boundaries + [100], "is_quantile": True, "n_samples": len(df), } @staticmethod def _build_narrative(bucket: BreadthBucket, divergence: float, ema_pct: float, ad_ratio: float) -> str: parts = [] if bucket == BreadthBucket.EXTREME: parts.append(f"全市场极度扩散({ema_pct:.0%}站上EMA20)") elif bucket == BreadthBucket.STRONG: parts.append("市场广度强势") elif bucket == BreadthBucket.NORMAL: parts.append("市场广度中性") elif bucket == BreadthBucket.WEAK: parts.append("市场广度疲弱") else: parts.append("市场广度恐慌") if divergence > 10: parts.append("资金集中于大市值(Top20>>Top50)") elif divergence < -10: parts.append("垃圾币狂欢(Top50>>Top20)") return ", ".join(parts)