refactor: p_oversold扫描器改读加工层——compute_market_filters读market_indicators+bias60/mcap_q/pe_q读stock_indicators+fetch_sector_momentum改查stock_sectors_em,不再fetch_tx_klines采集(使用层只读不采集,口径与回测一致)

deploy/profile-scripts/predictive_oversold_scanner.py

@@ -96,25 +96,19 @@ def load_market_state():
 
 
 def compute_market_filters():
-    """计算大盘门控：mkt_rsi / mkt_dd60（用上证指数K线）"""
-    # 000001 = 上证指数（fetch_tx_klines 自动加 sh 前缀；不能用带前缀的 sh000001）
-    klines = fetch_tx_klines("000001", datalen=80)
-    if not klines or len(klines) < 70:
+    """计算大盘门控：mkt_rsi / mkt_dd60（2026-08-12 改读 market_indicators 加工层预算值，
+    不再 fetch_tx_klines 自己拉指数K线算——使用层只读数据不采集，口径与回测一致）"""
+    try:
+        conn = sqlite3.connect(str(DB_PATH), timeout=5)
+        row = conn.execute(
+            "SELECT mkt_rsi, mkt_dd60, mkt_down_days FROM market_indicators ORDER BY date DESC LIMIT 1"
+        ).fetchone()
+        conn.close()
+        if not row:
+            return None
+        return {"mkt_rsi": row[0], "mkt_dd60": row[1], "down_days": row[2] or 0}
+    except Exception:
         return None
-    closes = [k["close"] for k in klines]
-    rsi_all = calc_rsi(closes)
-    mkt_rsi = rsi_all[-1] if rsi_all and rsi_all[-1] is not None else None
-    # 60日高点回撤
-    hi60 = max(closes[-60:]) if len(closes) >= 60 else max(closes)
-    mkt_dd60 = (closes[-1] - hi60) / hi60 * 100 if hi60 > 0 else 0
-    # 阴跌判定：连跌天数
-    down_days = 0
-    for i in range(len(closes) - 1, 0, -1):
-        if closes[i] < closes[i - 1]:
-            down_days += 1
-        else:
-            break
-    return {"mkt_rsi": mkt_rsi, "mkt_dd60": mkt_dd60, "down_days": down_days}
 
 
 def check_gate(mkt):
@@ -133,17 +127,12 @@ def check_gate(mkt):
     return True, "门控通过"
 
 
-def check_stock(code, name, mcap_q, pe_q, news3, sec_ret20, klines):
-    """个股条件检查（v5）：bias60 + 综合过滤"""
-    if not klines or len(klines) < 70:
-        return False, "K线不足"
-    closes = [k["close"] for k in klines]
-    ma60 = calc_ma(closes, 60)
-    m60 = ma60[-1]
-    close = closes[-1]
-    if not m60 or m60 <= 0:
-        return False, "MA60不可用"
-    bias60 = (close - m60) / m60 * 100
+def check_stock(code, name, mcap_q, pe_q, news3, sec_ret20, bias60):
+    """个股条件检查（v5）：bias60 + 综合过滤。
+    2026-08-12 改：bias60 由调用方从 stock_indicators 加工层预算值传入，
+    不再用 klines 现算 ma60——使用层只读数据不采集，口径与回测一致。"""
+    if bias60 is None:
+        return False, "bias60不可用"
     # 核心：深度超跌
     if bias60 >= OVERSOLD_CFG["bias60_max"]:
         return False, f"bias60={bias60:.1f}>-20，不够超跌"
@@ -220,27 +209,29 @@ def main():
             (code, SECTOR, f"{today}%")).fetchone()
         if r:
             continue
-        # 拉K线
-        klines = fetch_tx_klines(code)
-        if not klines:
+        # 2026-08-12 改：一次性读 stock_indicators 加工层预算值（bias60/mcap_q/pe_q），
+        # 替代 fetch_tx_klines + fetch_fundamentals + get_market_percentile——
+        # 使用层只读数据不采集，口径与回测完全一致（factor_engine 收盘后加工）
+        ind = conn.execute(
+            "SELECT bias60, mcap_q, pe_q FROM stock_indicators WHERE code=? ORDER BY date DESC LIMIT 1",
+            (code,)).fetchone()
+        if not ind:
             continue
-        # 基本面（PE/市值）
-        fund = fetch_fundamentals(code)
-        if not fund:
+        bias60, mcap_q, pe_q = ind
+        if bias60 is None or mcap_q is None or pe_q is None:
             continue
-        # 2026-08-11：真实分位（全市场 PE/市值分位，替代简化值）
-        mcap_q = get_market_percentile(code, "mcap_total")
-        pe_q = get_market_percentile(code, "pe")
-        if mcap_q is None or pe_q is None:
-            continue
-        # 2026-08-11：真实新闻数（3日）+ 行业20日动量
+        # 新闻3日 + 行业20日动量（读采集/加工层数据）
         news3 = fetch_news_count(code)
         sec_ret20 = fetch_sector_momentum(code)
 
-        ok_s, msg_s = check_stock(code, code, mcap_q, pe_q, news3, sec_ret20, klines)
+        ok_s, msg_s = check_stock(code, code, mcap_q, pe_q, news3, sec_ret20, bias60)
         if ok_s:
-            name = code
-            write_candidate(conn, code, name, msg_s, klines[-1]["close"])
+            # 入场价：stock_daily 最新收盘价
+            pr = conn.execute("SELECT close FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 1", (code,)).fetchone()
+            price = pr[0] if pr and pr[0] else 0
+            if price <= 0:
+                continue
+            write_candidate(conn, code, code, msg_s, price)
             hits += 1
             print(f"  🟢 {code} {msg_s}", flush=True)
         if hits >= 5:  # 单日限5（step39）
@@ -277,12 +268,13 @@ def get_market_percentile(code, field):
 
 
 def fetch_sector_momentum(code):
-    """计算行业20日动量（%）。从 stock_sectors 拿行业，再算行业指数20日涨跌。
-    返回 None 表示无行业数据。"""
+    """计算行业20日动量（%）。2026-08-12 改：从 stock_sectors_em（EM体系权威映射，
+    5061只/307行业，与回测 prepare_sector_context 对齐）拿行业，查 sector_index_daily
+    （sector_index_builder 加工层产物）算20日涨跌。返回 None 表示无行业数据。"""
     try:
         conn = sqlite3.connect(str(DB_PATH), timeout=5)
         sector = conn.execute(
-            "SELECT sector_name FROM stock_sectors WHERE code=? LIMIT 1", (code,)).fetchone()
+            "SELECT sector FROM stock_sectors_em WHERE code=? LIMIT 1", (code,)).fetchone()
         if not sector or not sector[0]:
             conn.close()
             return None