feat: S6叙事一致性检查+统一入口+容量60硬顶+提拔先入观察

- candidate_filter S6: 消息(个股+行业)x资金x技术叙事矩阵利好出货/三重打击=硬否决(dropped=1,不可被高分抵消) 共振做多+3/利空出尽+1/资金驱动+1/阴跌-1/行业利空-1 - promote: 信号一律先入'关注'(12维确认才升,消灭出生即买入); 容量60硬顶,RR<2的末位淘汰 - alphasift收编: 只写candidates不再直写自选(统一入口)
2026-07-24 12:09:30 +08:00
parent 481acfb18f
commit 309f074c51
3 changed files with 181 additions and 22 deletions
@@ -231,6 +231,118 @@ def stage4_capital_flow(code, name):

 # ── Stage 5: 基本面 ──

+# ── Stage 6: 叙事一致性检查（消息×资金×技术，2026-07-24 老爸重点要求）──
+# 否决项不可被其他 stage 高分抵消：叙事烂了评分再高也不进。
+
+def stage6_narrative(code, name):
+    """第六关：消息面×资金面×技术面 叙事一致性。
+    返回 (passed, score, detail)。score=-99 表示硬否决（利好出货/三重打击）。"""
+    conn = get_conn()
+    try:
+        # ── 1. 消息面（近3天，个股+行业）──
+        sector_name = ""
+        try:
+            r = conn.execute("SELECT sector_name FROM stock_sectors WHERE code=? LIMIT 1", (code,)).fetchone()
+            sector_name = r[0] if r else ""
+        except Exception:
+            pass
+        _NEG = ("高风险", "WATCH_HIGH", "紧急信号", "利空", "暴跌", "退市", "违约", "立案", "处罚", "亏损扩大")
+        _POS = ("利好", "大涨", "突破", "净流入", "中标", "订单", "预增", "回购")
+        stock_neg = stock_pos = 0
+        try:
+            rows = conn.execute(
+                "SELECT summary, overall_sentiment FROM signal_news "
+                "WHERE searched_stocks LIKE ? AND created_at >= datetime('now','-3 days') "
+                "ORDER BY id DESC LIMIT 5", (f"%{code}%",)).fetchall()
+            for summary, senti in rows:
+                t = f"{summary or ''}{senti or ''}"
+                if any(k in t for k in _NEG):
+                    stock_neg += 1
+                elif any(k in t for k in _POS):
+                    stock_pos += 1
+        except Exception:
+            pass
+        sector_neg = 0
+        if sector_name:
+            try:
+                rows = conn.execute(
+                    "SELECT summary, overall_sentiment FROM signal_news "
+                    "WHERE sector LIKE ? AND created_at >= datetime('now','-3 days') "
+                    "ORDER BY id DESC LIMIT 5", (f"%{sector_name}%",)).fetchall()
+                for summary, senti in rows:
+                    t = f"{summary or ''}{senti or ''}"
+                    if any(k in t for k in _NEG):
+                        sector_neg += 1
+            except Exception:
+                pass
+        if stock_neg > 0:
+            news = "利空"
+        elif stock_pos > 0:
+            news = "利好"
+        else:
+            news = "中性"
+
+        # ── 2. 资金面（主力净流入趋势，近5日）──
+        main_flow = 0.0
+        flow_trend = "中性"
+        try:
+            import json as _j
+            fr = conn.execute("SELECT cache_json FROM capital_flow_cache ORDER BY id DESC LIMIT 1").fetchone()
+            if fr and fr[0]:
+                fc = _j.loads(fr[0])
+                a = (fc.get("stocks", {}).get(code, {}) or {}).get("analysis", {})
+                main_flow = float(a.get("main_force", 0) or 0)
+                flow_trend = a.get("trend", "中性")
+        except Exception:
+            pass
+
+        # ── 3. 技术面（现价 vs 弱撑）──
+        above_support = True
+        try:
+            r = conn.execute(
+                "SELECT entry_low FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
+            lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
+            if r and lp and r[0] and lp[0]:
+                above_support = lp[0] >= r[0] * 0.98
+        except Exception:
+            pass
+
+        inflow = main_flow > 0
+        narrative = ""
+        # ── 4. 叙事矩阵 ──
+        if news == "利好" and not inflow:
+            narrative = f"利好出货(新闻利好{stock_pos}条但主力净流出{main_flow:.0f}万)"
+            conn.close()
+            return False, -99, narrative
+        if news == "利空" and not inflow and not above_support:
+            narrative = f"三重打击(利空{stock_neg}条+主力流出{main_flow:.0f}万+技术破位)"
+            conn.close()
+            return False, -99, narrative
+        if news == "利空" and inflow and above_support:
+            narrative = f"利空出尽(利空{stock_neg}条但主力流入+抗跌)"
+            score = 1
+        elif news == "利好" and inflow and above_support:
+            narrative = f"共振做多(利好{stock_pos}条+主力流入{main_flow:.0f}万+支撑上)"
+            score = 3
+        elif news == "中性" and inflow and above_support:
+            narrative = f"资金驱动(主力流入{main_flow:.0f}万)"
+            score = 1
+        elif news == "中性" and not inflow and not above_support:
+            narrative = "阴跌弱势"
+            score = -1
+        else:
+            narrative = f"中性(消息{news}/主力{main_flow:.0f}万)"
+            score = 0
+        if sector_neg > 0:
+            score -= 1
+            narrative += f"+行业利空{sector_neg}条"
+        conn.close()
+        return score >= 0, score, narrative
+    except Exception as e:
+        conn.close()
+        return True, 0, f"叙事检查异常放行: {str(e)[:60]}"
+
+
 def stage5_fundamental(code, name, price):
    """第五关：基本面
    从已有数据判断，不调外部API
@@ -284,10 +396,19 @@ def main():
    print(f"[FILTER] 待处理候选: {len(rows)}只", flush=True)
    
    stages = [(2, stage2_confirm, "多日K线"), (3, stage3_technical, "技术位"),
-              (4, stage4_capital_flow, "资金流"), (5, stage5_fundamental, "基本面")]
+              (4, stage4_capital_flow, "资金流"), (5, stage5_fundamental, "基本面"),
+              (6, stage6_narrative, "消息面叙事")]
+    
+    # S6 需要的列（幂等迁移）
+    for _col, _def in (("score_6th", "REAL"), ("pass_s6", "INTEGER")):
+        try:
+            conn.execute(f"ALTER TABLE candidates ADD COLUMN {_col} {_def}")
+        except Exception:
+            pass
    
    for code, name, reason in rows:
        current_score = 0
+        veto = False  # S6硬否决标记（利好出货/三重打击）
        print(f"  {code} {name}", flush=True)
        
        # 获取K线（多关需要）
@@ -336,15 +457,31 @@ def main():
                            (sscore, 1 if passed else 0, detail, code))
                log_candidate(conn, code, 5, passed, detail)
                print(f"    S5:{'✅' if passed else '❌'} {detail}", flush=True)
+            
+            elif stage_num == 6:
+                passed, sscore, detail = stage_fn(code, name)
+                if sscore <= -99:
+                    # 叙事硬否决：pass_s6=0, dropped=1, 综合分清零, 永不提拔
+                    veto = True
+                    conn.execute("UPDATE candidates SET score_6th=0, pass_s6=0, dropped=1, drop_reason=? WHERE code=?",
+                                (detail, code))
+                    log_candidate(conn, code, 6, False, detail)
+                    print(f"    S6:🚫否决 {detail}", flush=True)
+                else:
+                    conn.execute("UPDATE candidates SET score_6th=?, pass_s6=?, reason=? WHERE code=?",
+                                (sscore, 1 if passed else 0, detail, code))
+                    log_candidate(conn, code, 6, passed, detail)
+                    print(f"    S6:{'✅' if passed else '❌'} {detail}", flush=True)
        
-        # 计算综合评分
+        # 计算综合评分（S6否决则清零）
        s2 = conn.execute("SELECT score_2nd FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
        s3 = conn.execute("SELECT score_3rd FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
        s4 = conn.execute("SELECT score_4th FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
        s5 = conn.execute("SELECT score_5th FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
-        final = current_score + s2 + s3 + s4 + s5
-        conn.execute("UPDATE candidates SET score_final=?, pass_final=1 WHERE code=?",
-                    (final, code))
+        s6 = conn.execute("SELECT score_6th FROM candidates WHERE code=?", (code,)).fetchone()[0] or 0
+        final = 0 if veto else (current_score + s2 + s3 + s4 + s5 + s6)
+        conn.execute("UPDATE candidates SET score_final=?, pass_final=? WHERE code=?",
+                    (final, 0 if veto else 1, code))
    
    conn.commit()
    conn.close()
@@ -209,29 +209,29 @@ def run_all(strategies_str, market, max_results, dry_run=False):
        print("\n[DRY RUN] 未写入")
        return
    
-    # 写入 DB
+    # 写入 DB（2026-07-24 老爸"统一入口"：只写 candidates，不许直写自选。
+    # 后续由 promote_candidates 的 score>=7+RR>=2+S6叙事闸 统一提拔）
    try:
        sys.path.insert(0, str(MOFIN_DATA.parent))
-        from mofin_db import get_conn, write_watchlist_stock
+        from mofin_db import get_conn
        conn = get_conn()
        for s in new_stocks:
-            s.setdefault('currency', 'CNY')
-            write_watchlist_stock(conn, s)
+            sd = s.get("source_detail", {})
+            conn.execute(
+                "INSERT INTO candidates (code, name, sector, reason, entry_range, stop_loss, target, created_at) "
+                "VALUES (?,?,?,?,?,?,?,datetime('now','localtime')) "
+                "ON CONFLICT(code) DO UPDATE SET "
+                "name=excluded.name, sector=excluded.sector, reason=excluded.reason, "
+                "entry_range=excluded.entry_range, stop_loss=excluded.stop_loss, target=excluded.target",
+                (s["code"], s["name"], "alpha_sift",
+                 f"{s.get('notes','')}",
+                 "", 0, 0))
+        conn.commit()
        conn.close()
-        print(f"\n已写入 {len(new_stocks)} 只到 DB watchlist_stocks")
+        print(f"\n已写入 {len(new_stocks)} 只到 candidates（等待 candidate_filter 多级过滤+promote 统一提拔）")
    except Exception as e:
        print(f"WARN: DB写入失败: {e}")
        return
-    
-    # 策略生成
-    print("\n调用 regenerate_all()...")
-    try:
-        sys.path.insert(0, str(MOFIN_DATA.parent))
-        from strategy_lifecycle import regenerate_all
-        r = regenerate_all(stdout=True)
-        if r: print(f"完成: {r.get('ok',0)}/{r.get('total',0)} 只策略已生成")
-    except Exception as e:
-        print(f"WARN: {e}")


 def list_strategies():
@@ -84,9 +84,9 @@ def main():
                print(f"  ⏭ {code} {name} RR={_rr:.2f}<2.0，不入自选")
                continue
        
-        # 构建策略
+        # 构建策略（2026-07-24 老爸：提拔不直接给"买入"——先入观察，12维确认后再升）
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-        timing_signal = "买入" if score >= 7 else "关注"
+        timing_signal = "关注"
        price_est = (el + eh) / 2 if el > 0 and eh > 0 else 0
        reason_text = []
        if el > 0: reason_text.append(f"买{el}~{eh}")
@@ -98,6 +98,28 @@ def main():
        reason_text.append(f"评分{score}")
        action = " | ".join(reason_text) if reason_text else f"市场扫描发现(评分{score})"
        
+        # ── 容量闸：自选上限60只，超出时删综合分最弱的（RR低优先）──
+        MAX_WATCH = 60
+        wl_count = conn.execute(
+            "SELECT COUNT(*) FROM holding_strategies WHERE status='active' AND decision_type='自选策略'").fetchone()[0]
+        if wl_count >= MAX_WATCH:
+            weakest = conn.execute("""
+                SELECT code, name, COALESCE(rr_ratio,0) as rr FROM holding_strategies
+                WHERE status='active' AND decision_type='自选策略'
+                ORDER BY COALESCE(rr_ratio,0) ASC, updated_at ASC LIMIT 1""").fetchone()
+            if weakest and (weakest[2] or 0) < 2.0:
+                conn.execute(
+                    "INSERT INTO watchlist_log (code, name, event, reason, old_signal, new_signal, price) "
+                    "VALUES (?,?,?,?,?,?,?)",
+                    (weakest[0], weakest[1] or "", "exit", f"容量{MAX_WATCH}淘汰为新标的{code}腾位", "", "已删除", 0))
+                conn.execute(
+                    "DELETE FROM holding_strategies WHERE code=? AND status='active' AND decision_type='自选策略'",
+                    (weakest[0],))
+                print(f"  🔴 容量淘汰: {weakest[0]} {weakest[1]} (RR={weakest[2]})", flush=True)
+            else:
+                print(f"  ⏭ 自选已满{MAX_WATCH}且现有标的均RR>=2.0，{code}暂缓提拔", flush=True)
+                continue
+
        cur = conn.execute("""
            INSERT OR IGNORE INTO holding_strategies 
            (code, name, price, entry_low, entry_high, stop_loss, take_profit,