diff --git a/deploy/profile-scripts/candidate_filter.py b/deploy/profile-scripts/candidate_filter.py index 4a48c068..6438ae17 100644 --- a/deploy/profile-scripts/candidate_filter.py +++ b/deploy/profile-scripts/candidate_filter.py @@ -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() diff --git a/deploy/profile-scripts/mo_alphasift_bridge.py b/deploy/profile-scripts/mo_alphasift_bridge.py index fe4989bf..7cbc3343 100644 --- a/deploy/profile-scripts/mo_alphasift_bridge.py +++ b/deploy/profile-scripts/mo_alphasift_bridge.py @@ -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(): diff --git a/deploy/profile-scripts/promote_candidates.py b/deploy/profile-scripts/promote_candidates.py index cdb17daf..426f51cc 100644 --- a/deploy/profile-scripts/promote_candidates.py +++ b/deploy/profile-scripts/promote_candidates.py @@ -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,