""" scoring/price_structure.py — Price Structure Score (OHLCV-only). Three sub-dimensions: 1. Trend Strength (40%): EMA alignment + ADX 2. Volatility Compression (30%): ATR + BB width 3. Momentum (30%): ROC + consecutive candles This module works with zero external dependencies — just OHLCV data. """ from datetime import date as Date import sqlite3 import math import numpy as np import pandas as pd from .base import BaseScorer from .constants import ( ADX_TREND_THRESHOLD, ADX_STRONG_THRESHOLD, BB_COMPRESSION_LOW, BB_COMPRESSION_HIGH, ROC_STRONG_BULLISH, ROC_STRONG_BEARISH, CONSECUTIVE_CANDLES_SIGNAL, ) from models import FactorScore, PriceStructureScore, MacroDirection from config import config class PriceStructureScorer(BaseScorer): """Scores market structure from OHLCV data alone.""" def compute(self, target_date: Date) -> PriceStructureScore: conn = self.get_connection() try: df = self._load_ohlcv(conn, str(target_date), lookback=120) if df.empty: return PriceStructureScore( name="Price Structure", score=50.0, label="No Data", ) trend = self._score_trend_strength(df) vol_comp = self._score_volatility_compression(df) momentum = self._score_momentum(df) # Weighted aggregate score = trend * 0.40 + vol_comp * 0.30 + momentum * 0.30 # Determine direction if trend > 60: direction = MacroDirection.BULLISH elif trend < 40: direction = MacroDirection.BEARISH else: direction = MacroDirection.NEUTRAL # Build narrative latest = df.iloc[-1] narrative = self._build_narrative(trend, vol_comp, momentum, latest) return PriceStructureScore( name="Price Structure", score=round(score, 1), label=self._label(score), direction=direction, trend_strength=round(trend, 1), volatility_compression=round(vol_comp, 1), momentum=round(momentum, 1), sub_scores={ "trend_strength": round(trend, 1), "volatility_compression": round(vol_comp, 1), "momentum": round(momentum, 1), }, narrative=narrative, ) finally: conn.close() def _load_ohlcv(self, conn: sqlite3.Connection, date_str: str, lookback: int = 120) -> pd.DataFrame: """Load OHLCV data up to target_date.""" df = pd.read_sql_query( "SELECT * FROM ohlcv_daily WHERE date <= ? ORDER BY date DESC LIMIT ?", conn, params=(date_str, lookback) ) if df.empty: return df return df.sort_values("date").reset_index(drop=True) def _score_trend_strength(self, df: pd.DataFrame) -> float: """Score trend based on EMA alignment and ADX.""" latest = df.iloc[-1] # EMA alignment ema20 = latest.get("ema20") ema60 = latest.get("ema60") ema120 = latest.get("ema120") ema_score = 50.0 if ema20 and ema60 and ema120 and not pd.isna(ema20) and not pd.isna(ema60) and not pd.isna(ema120): alignments = 0 if ema20 > ema60: alignments += 1 if ema60 > ema120: alignments += 1 if ema20 > ema120: alignments += 1 # Distance from EMAs close = float(latest["close"]) ema20_dist = abs(close - ema20) / ema20 * 100 if ema20 else 0 if alignments == 3: ema_score = 80 + min(ema20_dist, 15) # strong bullish alignment elif alignments == 0: ema_score = 20 - min(ema20_dist, 15) # strong bearish alignment elif alignments == 2: ema_score = 65 else: ema_score = 35 # ADX adx = latest.get("adx_14") adx_score = 50.0 if adx and not pd.isna(adx): if adx > ADX_STRONG_THRESHOLD: adx_score = 85 elif adx > ADX_TREND_THRESHOLD: adx_score = 65 + (adx - ADX_TREND_THRESHOLD) / (ADX_STRONG_THRESHOLD - ADX_TREND_THRESHOLD) * 20 else: adx_score = 50 - (ADX_TREND_THRESHOLD - adx) / ADX_TREND_THRESHOLD * 30 return ema_score * 0.55 + adx_score * 0.45 def _score_volatility_compression(self, df: pd.DataFrame) -> float: """Score volatility compression — expansion = high, compression = low-mid.""" latest = df.iloc[-1] bb_width = latest.get("bb_width") if not bb_width or pd.isna(bb_width) or len(df) < 20: return 50.0 # BB width relative to 20d average recent_bb = df["bb_width"].dropna().tail(20) if len(recent_bb) < 10: return 50.0 bb_avg = recent_bb.mean() bb_ratio = bb_width / bb_avg if bb_avg > 0 else 1.0 if bb_ratio < BB_COMPRESSION_LOW: # Compression → potential breakout, neutral-bullish return 45 + (BB_COMPRESSION_LOW - bb_ratio) * 30 elif bb_ratio > BB_COMPRESSION_HIGH: # Expansion → trending or chaotic return 75 + min((bb_ratio - BB_COMPRESSION_HIGH) * 20, 20) else: # Normal return 55 def _score_momentum(self, df: pd.DataFrame) -> float: """Score momentum using ROC and consecutive candles.""" if len(df) < 10: return 50.0 closes = df["close"].astype(float) latest = float(closes.iloc[-1]) # ROC (5-bar) if len(closes) >= 6: roc5 = (closes.iloc[-1] - closes.iloc[-6]) / closes.iloc[-6] * 100 else: roc5 = 0 # ROC (10-bar) if len(closes) >= 11: roc10 = (closes.iloc[-1] - closes.iloc[-11]) / closes.iloc[-11] * 100 else: roc10 = 0 # ROC (20-bar) if len(closes) >= 21: roc20 = (closes.iloc[-1] - closes.iloc[-21]) / closes.iloc[-21] * 100 else: roc20 = 0 # Score ROC: map to 0-100 def roc_to_score(roc, scale=15): return 50 + np.clip(roc / scale * 50, -50, 50) roc_score = roc_to_score(roc5, 10) * 0.4 + roc_to_score(roc10, 15) * 0.35 + roc_to_score(roc20, 20) * 0.25 # Consecutive candle direction consec_score = 50.0 consec_up = 0 consec_down = 0 for i in range(len(closes) - 1, max(0, len(closes) - 10), -1): if closes.iloc[i] > closes.iloc[i - 1]: consec_up += 1 consec_down = 0 elif closes.iloc[i] < closes.iloc[i - 1]: consec_down += 1 consec_up = 0 else: break if consec_up >= CONSECUTIVE_CANDLES_SIGNAL: consec_score = 70 + min(consec_up * 5, 25) elif consec_down >= CONSECUTIVE_CANDLES_SIGNAL: consec_score = 30 - min(consec_down * 5, 25) return roc_score * 0.70 + consec_score * 0.30 def _build_narrative(self, trend: float, vol: float, momentum: float, latest: pd.Series) -> str: parts = [] if trend > 65: parts.append("EMA多头排列+ADX趋势明确") elif trend > 50: parts.append("趋势温和偏多") elif trend < 35: parts.append("EMA空头排列+ADX趋势明确") elif trend < 50: parts.append("趋势温和偏空") else: parts.append("趋势中性") if vol > 70: parts.append("波动率扩张") elif vol < 45: parts.append("波动率压缩(突破前兆)") if momentum > 65: parts.append("动量强劲") elif momentum < 35: parts.append("动量疲弱") return ", ".join(parts) if parts else "中性" @staticmethod def _label(score: float) -> str: if score >= 75: return "Strong Bullish Structure" elif score >= 60: return "Bullish Structure" elif score >= 40: return "Neutral Structure" elif score >= 25: return "Bearish Structure" return "Weak Bearish Structure"