fix: 52只缺失策略补全+inactive概念清除+self_repair噪音过滤+dedup

archive/20260728-dedup/candidate_filter.py

@@ -0,0 +1,354 @@
+#!/usr/bin/env python3
+"""candidate_filter.py — 候选股多级过滤管道
+
+从 candidates 表读取未过滤的候选，逐级执行过滤：
+  Stage 2: 多日K线确认（量价连续性）
+  Stage 3: 技术位分析（MA位置）
+  Stage 4: 资金性质（大单流向）
+  Stage 5: 基本面（PE/PB/行业）
+
+用法: python3 candidate_filter.py [--stage 2|3|4|5] [--code XXXXXX]
+"""
+import sys, json, urllib.request, sqlite3, re, time
+from pathlib import Path
+from datetime import datetime
+
+DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
+UA = "Mozilla/5.0"
+
+def get_conn():
+    c = sqlite3.connect(str(DB_PATH), timeout=30)
+    c.execute("PRAGMA busy_timeout=30000")
+    return c
+
+def log_candidate(conn, code, stage, passed, detail):
+    """记录过滤日志"""
+    conn.execute(
+        "UPDATE candidates SET log = COALESCE(log, '[]')"
+    )
+    # SQLite JSON操作
+    existing = conn.execute("SELECT log FROM candidates WHERE code=?", (code,)).fetchone()
+    if existing and existing[0]:
+        try:
+            logs = json.loads(existing[0])
+        except:
+            logs = []
+    else:
+        logs = []
+    logs.append({"stage": stage, "passed": passed, "detail": detail, "time": datetime.now().strftime("%m-%d %H:%M")})
+    conn.execute("UPDATE candidates SET log=? WHERE code=?", (json.dumps(logs, ensure_ascii=False), code))
+
+
+# ── Stage 2: 多日K线确认 ──
+
+def fetch_daily_klines(code):
+    """拉取近10日日K线（Sina 240分钟线=日K）"""
+    raw = str(code).strip()
+    if raw.startswith(("6", "9")):
+        prefix = "sh"
+    elif raw.startswith(("0", "3")):
+        prefix = "sz"
+    else:
+        return None
+    
+    import subprocess as _sp, json as _json
+    url = f"http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol={prefix}{raw}&scale=240&ma=5&datalen=10"
+    try:
+        r = _sp.run(["curl", "-s", "--noproxy", "*", url], capture_output=True, timeout=10)
+        data = _json.loads(r.stdout)
+        if not data:
+            return None
+        result = []
+        for k in data:
+            result.append({
+                "date": k.get("day", "")[:10],
+                "open": float(k["open"]),
+                "close": float(k["close"]),
+                "high": float(k["high"]),
+                "low": float(k["low"]),
+                "volume": int(k["volume"]),
+                "price": float(k["close"]),
+                "change_pct": 0,
+            })
+        # 计算涨跌幅
+        for i in range(1, len(result)):
+            prev = result[i-1]["close"]
+            if prev > 0:
+                result[i]["change_pct"] = (result[i]["close"] / prev - 1) * 100
+        return result
+    except Exception as e:
+        return None
+        return None
+
+
+def stage2_confirm(code, name, klines):
+    """第二关：多日K线确认
+    检查：多日量价配合、建仓特征
+    """
+    if not klines or len(klines) < 3:
+        return False, 0, "K线不足3日"
+    
+    recent = klines[-5:]  # 最近5日
+    score = 0
+    checks = []
+    
+    # 1. 成交量连续递增
+    vols = [k["volume"] for k in recent]
+    vol_rising = sum(1 for i in range(len(vols)-1) if vols[i] < vols[i+1])
+    if vol_rising >= 3:
+        score += 2
+        checks.append(f"量增{vol_rising}/4日")
+    elif vol_rising >= 2:
+        score += 1
+        checks.append(f"量微增{vol_rising}/4日")
+    
+    # 2. 涨放量、跌缩量
+    up_vol = sum(k["volume"] for k in recent if k["change_pct"] >= 0)
+    down_vol = sum(k["volume"] for k in recent if k["change_pct"] < 0)
+    if down_vol > 0 and up_vol / down_vol > 1.5:
+        score += 2
+        checks.append(f"涨量/跌量={up_vol/down_vol:.1f}")
+    elif down_vol > 0 and up_vol / down_vol > 1:
+        score += 1
+    
+    # 3. 价格趋势
+    closes = [k["close"] for k in recent]
+    up_days = sum(1 for i in range(1, len(closes)) if closes[i] > closes[i-1])
+    if up_days >= 3:
+        score += 2
+        checks.append(f"涨{up_days}/4日")
+    elif up_days >= 2:
+        score += 1
+    
+    # 4. 无异常放量（单日>3倍均量=可能出货）
+    avg_vol = sum(vols) / len(vols) if vols else 1
+    max_ratio = max(v / avg_vol for v in vols) if avg_vol > 0 else 1
+    if max_ratio < 2.5:
+        score += 1
+    else:
+        checks.append(f"异常量{max_ratio:.0f}倍")
+    
+    passed = score >= 4
+    detail = f"评分{score}/7 | {'; '.join(checks)}"
+    return passed, score, detail
+
+
+# ── Stage 3: 技术位分析 ──
+
+def stage3_technical(code, name, klines):
+    """第三关：技术位（当日数据估算）"""
+    if not klines or len(klines) == 0:
+        return False, 0, "无数据"
+    
+    today = klines[-1]
+    price = today.get("price", 0)
+    high = today.get("high", 0)
+    low = today.get("low", 0)
+    
+    score = 0
+    checks = []
+    
+    if price <= 0:
+        return False, 0, "价格无效"
+    
+    # 日内位置（在高低点中下段还有空间）
+    if high > low:
+        pos = (price - low) / (high - low)
+        if pos < 0.7:
+            score += 1
+            checks.append(f"日内位置{pos:.0%}")
+    
+    # 有明确支撑（今日低点作为参考支撑）
+    if low > 0 and price > low:
+        score += 1
+        checks.append(f"支撑{low:.2f}")
+    
+    # 有上涨空间（今日高点作为参考阻力）
+    if high > price:
+        upside = (high / price - 1) * 100
+        if upside > 2:
+            score += 1
+            checks.append(f"空间{upside:.0f}%")
+    
+    passed = score >= 2
+    return passed, score, "; ".join(checks) if checks else "基础通过"
+
+
+# ── Stage 4: 资金性质分析 ──
+
+def stage4_capital_flow(code, name):
+    """第四关：资金性质（从腾讯实时行情提取外盘/内盘比）"""
+    raw = str(code).strip()
+    if raw.startswith(("6", "9")):
+        prefix = "sh"
+    elif raw.startswith(("0", "3")):
+        prefix = "sz"
+    else:
+        return False, 0, "非A股"
+    
+    import subprocess as _sp
+    url = f"http://qt.gtimg.cn/q={prefix}{raw}"
+    try:
+        r = _sp.run(["curl", "-s", url], capture_output=True, timeout=10)
+        text = r.stdout.decode("gbk", errors="ignore")
+        parts = text.split("~")
+        if len(parts) < 40:
+            return False, 0, "数据不足"
+        
+        # 腾讯字段：[7]=外盘(主动买,股),[8]=内盘(主动卖,股)
+        try:
+            outer = int(float(parts[7])) if parts[7] else 0  # 外盘
+            inner = int(float(parts[8])) if parts[8] else 0  # 内盘
+        except:
+            return False, 0, "解析失败"
+        
+        if outer <= 0 or inner <= 0:
+            return False, 0, "无盘口数据"
+        
+        score = 0
+        ratio = outer / inner if inner > 0 else 1
+        checks = []
+        
+        if ratio > 1.3:
+            score += 2
+            checks.append(f"外/内={ratio:.2f}")
+        elif ratio > 1.0:
+            score += 1
+            checks.append(f"买稍强{ratio:.2f}")
+        else:
+            checks.append(f"卖稍强{ratio:.2f}")
+        
+        # 绝对量也说明资金活跃度
+        total = outer + inner
+        if total > 50000000:  # >5000万股
+            score += 1
+            checks.append(f"活跃{total/10000:.0f}万")
+        
+        return score >= 1, score, "; ".join(checks)
+    except:
+        return False, 0, "接口失败"
+
+
+# ── Stage 5: 基本面 ──
+
+def stage5_fundamental(code, name, price):
+    """第五关：基本面
+    从已有数据判断，不调外部API
+    """
+    conn = get_conn()
+    score = 0
+    checks = []
+    
+    # PE（从stocks表或live_prices）
+    r = conn.execute("SELECT 1 FROM holdings WHERE code=? AND is_active=1", (code,)).fetchone()
+    is_holding = r is not None
+    if is_holding:
+        checks.append("已持仓")
+    else:
+        score += 1  # 新标的加分
+    
+    # 检查是否已被其他候选覆盖
+    r2 = conn.execute("SELECT code FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
+    if r2:
+        checks.append("已有策略")
+    else:
+        score += 1
+    
+    conn.close()
+    return score >= 1, score, "; ".join(checks) if checks else "新标的"
+
+
+# ── 主流程 ──
+
+def main():
+    stage_filter = None
+    single_code = None
+    for i, arg in enumerate(sys.argv[1:]):
+        if arg == "--stage" and i+1 < len(sys.argv):
+            stage_filter = int(sys.argv[i+2])
+        if arg == "--code" and i+1 < len(sys.argv):
+            single_code = sys.argv[i+2]
+    
+    conn = get_conn()
+    
+    # 读待过滤的候选
+    query = "SELECT code, name, reason FROM candidates WHERE 1=1"
+    params = []
+    if single_code:
+        query += " AND code=?"
+        params.append(single_code)
+    else:
+        query += " AND (pass_final IS NULL OR pass_final=0)"
+    
+    rows = conn.execute(query, params).fetchall()
+    print(f"[FILTER] 待处理候选: {len(rows)}只", flush=True)
+    
+    stages = [(2, stage2_confirm, "多日K线"), (3, stage3_technical, "技术位"),
+              (4, stage4_capital_flow, "资金流"), (5, stage5_fundamental, "基本面")]
+    
+    for code, name, reason in rows:
+        current_score = 0
+        print(f"  {code} {name}", flush=True)
+        
+        # 获取K线（多关需要）
+        klines = None
+        
+        for stage_num, stage_fn, stage_name in stages:
+            if stage_filter and stage_num != stage_filter:
+                continue
+            
+            # 检查是否已通过此关
+            col = f"pass_s{stage_num}"
+            existing = conn.execute(f"SELECT {col} FROM candidates WHERE code=?", (code,)).fetchone()
+            if existing and existing[0]:
+                continue
+            
+            if stage_num in (2, 3) and klines is None:
+                klines = fetch_daily_klines(code)
+            
+            if stage_num == 2:
+                passed, sscore, detail = stage_fn(code, name, klines)
+                conn.execute("UPDATE candidates SET score_2nd=?, pass_s2=?, reason=? WHERE code=?",
+                            (sscore, 1 if passed else 0, detail, code))
+                log_candidate(conn, code, 2, passed, detail)
+                print(f"    S2:{'✅' if passed else '❌'} {detail}", flush=True)
+            
+            elif stage_num == 3:
+                passed, sscore, detail = stage_fn(code, name, klines)
+                conn.execute("UPDATE candidates SET score_3rd=?, pass_s3=?, reason=? WHERE code=?",
+                            (sscore, 1 if passed else 0, detail, code))
+                log_candidate(conn, code, 3, passed, detail)
+                print(f"    S3:{'✅' if passed else '❌'} {detail}", flush=True)
+            
+            elif stage_num == 4:
+                passed, sscore, detail = stage_fn(code, name)
+                conn.execute("UPDATE candidates SET score_4th=?, pass_s4=?, reason=? WHERE code=?",
+                            (sscore, 1 if passed else 0, detail, code))
+                log_candidate(conn, code, 4, passed, detail)
+                print(f"    S4:{'✅' if passed else '❌'} {detail}", flush=True)
+            
+            elif stage_num == 5:
+                price = 0  # 从live_prices获取
+                r = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
+                if r: price = r[0]
+                passed, sscore, detail = stage_fn(code, name, price)
+                conn.execute("UPDATE candidates SET score_5th=?, pass_s5=?, reason=? WHERE code=?",
+                            (sscore, 1 if passed else 0, detail, code))
+                log_candidate(conn, code, 5, passed, detail)
+                print(f"    S5:{'✅' if passed else '❌'} {detail}", flush=True)
+        
+        # 计算综合评分
+        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))
+    
+    conn.commit()
+    conn.close()
+    print(f"[FILTER] 完成", flush=True)
+
+if __name__ == "__main__":
+    main()

archive/20260728-dedup/mofin_collect.py

@@ -0,0 +1,272 @@
+#!/usr/bin/env python3
+"""mofin_collect.py — MoFin 数据采集链
+
+每轮盯盘 cron 前运行，顺序执行：
+  0. 策略新鲜度强制检查（代码级约束：推荐前必须先检查策略是否最新）
+  1. market_watch — 拉90个行业板块数据（9:30前跳过，市场未开）
+  2. trend_detector — 检测17种信号（依赖板块数据，同跳）
+  3. mofin_news — 搜新闻+小果分析
+  4. stock_quote — 所有持仓最新行情（CRITICAL: LLM唯一价格源）
+"""
+
+import subprocess, sys, time, json
+from pathlib import Path
+from datetime import datetime
+from mo_data import get_price, get_prices_batch
+
+BASE = Path(__file__).parent.parent if "hermes" in str(Path(__file__).resolve()) else Path(__file__).parent
+
+now = datetime.now()
+market_open = (now.hour >= 9 and now.minute >= 30) or now.hour >= 10
+
+# ── 步骤0: 策略新鲜度强制检查（代码级约束）──
+# 在LLM看到任何数据前，先确保所有持仓策略是新鲜的
+# 策略过期(>4h未更新) → 创建TODO触发重评 + 注入STALE标记到LLM context
+# LLM看到STALE标记：禁止对该股给出任何操作建议
+print("--- strategy_freshness_check ---", flush=True)
+try:
+    sys.path.insert(0, str(BASE))
+    from mofin_db import get_conn
+    conn = get_conn()
+    cur = conn.cursor()
+    # 读所有活跃持仓及其最新策略更新时间
+    rows = cur.execute("""
+        SELECT h.code, h.name, h.price, h.cost, h.shares, h.position_pct,
+               hs.stop_loss, hs.take_profit, hs.entry_low, hs.entry_high,
+               hs.created_at, hs.action
+        FROM holdings h
+        LEFT JOIN (
+            SELECT code, stop_loss, take_profit, entry_low, entry_high,
+                   created_at, action,
+                   ROW_NUMBER() OVER (PARTITION BY code ORDER BY id DESC) AS rn
+            FROM holding_strategies
+        ) hs ON h.code = hs.code AND hs.rn = 1
+        WHERE h.is_active = 1
+        ORDER BY h.code
+    """).fetchall()
+    
+    stale_count = 0
+    fresh_count = 0
+    no_strategy_count = 0
+    
+    for r in rows:
+        code = r["code"]
+        name = r["name"]
+        last_update = r["created_at"]
+        has_strategy = last_update is not None
+        
+        if has_strategy:
+            try:
+                last_dt = datetime.fromisoformat(last_update)
+                hours_stale = (now - last_dt).total_seconds() / 3600
+            except:
+                hours_stale = 999
+            
+            if hours_stale > 4:
+                stale_count += 1
+                # 强制代码级约束：立即执行重评，不等TODO异步处理
+                # 在LLM看到数据前，策略必须是新鲜的
+                try:
+                    from strategy_lifecycle import reassess_with_context
+                    result = reassess_with_context(
+                        code, name, r["price"],
+                        r["cost"] if r["cost"] else 0, r["shares"] if r["shares"] else 0,
+                        r["action"] or ""
+                    )
+                    if result and result.get("action"):
+                        print(f"  🔄 FORCE_REASSESS {code} {name}: {hours_stale:.0f}h过期→已立即重评→{result['action'][:60]}", flush=True)
+                        # 写strategy_evaluations（独立短连接，防锁冲突）
+                        try:
+                            eval_conn = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"), timeout=10)
+                            eval_conn.execute("""
+                                INSERT INTO strategy_evaluations 
+                                (code, eval_type, status, new_stop_loss, new_tp, reason, created_at)
+                                VALUES (?, 'reassess', 'completed', ?, ?, ?, ?)
+                            """, (
+                                code,
+                                result.get("stop_loss"),
+                                result.get("take_profit"),
+                                f"{result.get('action','')} RR={result.get('rr_ratio','?')} 信号={result.get('timing_signal','')}",
+                                now.isoformat()
+                            ))
+                            eval_conn.commit()
+                            eval_conn.close()
+                        except Exception:
+                            pass
+                    else:
+                        print(f"  ⚠️ FORCE_REASSESS {code} {name}: 重评返回空结果", flush=True)
+                except Exception as e:
+                    print(f"  ❌ FORCE_REASSESS {code} {name} 失败: {e}", flush=True)
+                    # 创建TODO作为兜底
+                    todo_sql = """
+                        INSERT OR IGNORE INTO todos 
+                        (title, code, fix_action, source, priority, status, created_at)
+                        VALUES (?, ?, 'reassess_strategy', 'freshness_check', 'high', 'pending', ?)
+                    """
+                    cur.execute(todo_sql, (
+                        f"策略过期需重评: {code} {name} ({hours_stale:.0f}h未更新)",
+                        code,
+                        now.isoformat()
+                    ))
+                print(f"  ⚠️ STALE {code} {name}: {hours_stale:.0f}h未更新 → 已强制重评", flush=True)
+            else:
+                fresh_count += 1
+                print(f"  ✅ FRESH {code} {name}: {hours_stale:.1f}h前更新", flush=True)
+        else:
+            no_strategy_count += 1
+            # 无策略→立即执行重评创建策略（代码级约束：不允许无策略就输出建议）
+            try:
+                from strategy_lifecycle import reassess_with_context
+                result = reassess_with_context(
+                    code, name, r["price"],
+                    r["cost"] if r["cost"] else 0, r["shares"] if r["shares"] else 0,
+                    ""
+                )
+                if result and result.get("action"):
+                    print(f"  🔄 CREATE_STRATEGY {code} {name}: 无策略→已创建→{result['action'][:60]}", flush=True)
+                    try:
+                        eval_conn = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"), timeout=10)
+                        eval_conn.execute("""
+                            INSERT INTO strategy_evaluations 
+                            (code, eval_type, status, new_stop_loss, new_tp, reason, created_at)
+                            VALUES (?, 'reassess', 'completed', ?, ?, ?, ?)
+                        """, (
+                            code,
+                            result.get("stop_loss"),
+                            result.get("take_profit"),
+                            f"{result.get('action','')} RR={result.get('rr_ratio','?')} 信号={result.get('timing_signal','')}",
+                            now.isoformat()
+                        ))
+                        eval_conn.commit()
+                        eval_conn.close()
+                    except Exception:
+                        pass
+                else:
+                    print(f"  ⚠️ CREATE_STRATEGY {code} {name}: 重评返回空", flush=True)
+            except Exception as e:
+                print(f"  ❌ CREATE_STRATEGY {code} {name} 失败: {e}", flush=True)
+
+    conn.commit()
+
+    # === 自选股策略检查 + 强制重评 ===
+    wl_fresh = 0
+    wl_stale = 0
+    wl_error = 0
+    try:
+        for wr in conn.execute("SELECT code, name, price, entry_low, entry_high, stop_loss FROM watchlist_stocks WHERE is_active=1"):
+            code = wr["code"]
+            name = wr["name"]
+            wl_price = wr["price"] or 0
+            # 自选股price可能为0（新加入未更新），从实时API获取
+            if wl_price <= 0:
+                try:
+                    p, _ = get_price(code)
+                    if p and p > 0:
+                        wl_price = p
+                except Exception:
+                    pass
+            # 自选股无cost/shares，传0
+            try:
+                from strategy_lifecycle import reassess_with_context
+                result = reassess_with_context(
+                    code, name, wl_price,
+                    0, 0, ""
+                )
+                if result and result.get("action"):
+                    wl_stale += 1
+                    print(f"  📋 WATCHLIST_REASSESS {code} {name}: →{result['action'][:60]}", flush=True)
+                    try:
+                        eval_conn = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"), timeout=10)
+                        eval_conn.execute("""
+                            INSERT INTO strategy_evaluations 
+                            (code, eval_type, status, new_stop_loss, new_tp, reason, created_at)
+                            VALUES (?, 'reassess', 'completed', ?, ?, ?, ?)
+                        """, (
+                            code,
+                            result.get("stop_loss"),
+                            result.get("take_profit"),
+                            f"自选:{result.get('action','')} RR={result.get('rr_ratio','?')} 信号={result.get('timing_signal','')}",
+                            now.isoformat()
+                        ))
+                        eval_conn.commit()
+                        eval_conn.close()
+                    except Exception:
+                        pass
+                else:
+                    wl_fresh += 1
+            except Exception as e:
+                wl_error += 1
+                print(f"  ❌ WATCHLIST_REASSESS {code} {name} 失败: {e}", flush=True)
+    except Exception as e:
+        print(f"  ⚠️ 自选股检查跳过: {e}", flush=True)
+
+    conn.close()
+
+    total = len(rows)
+    wl_total = wl_fresh + wl_stale + wl_error
+    print(f"策略检查完成: {total}只持仓({fresh_count}新鲜/{stale_count}过期/{no_strategy_count}无策略) + {wl_total}只自选({wl_fresh}无需/{wl_stale}已重评/{wl_error}失败)", flush=True)
+    if stale_count > 0 or no_strategy_count > 0 or wl_stale > 0:
+        print(f"⚠️ 重评完成: {stale_count + no_strategy_count}只已强制刷新, LLM可基于最新策略给出建议", flush=True)
+except Exception as e:
+    print(f"WARN: strategy_freshness_check跳过 ({e})", flush=True)
+
+# 步骤1-3: 行业/新闻数据
+SCRIPTS = []
+if market_open:
+    SCRIPTS.append(("market_watch.py", 60))
+    SCRIPTS.append(("trend_detector.py", 60))
+else:
+    print(f"[{now.strftime('%H:%M')}] 市场未开盘(9:30)，跳过板块采集", flush=True)
+
+SCRIPTS.append(("mofin_news.py", 50))
+
+for script, timeout in SCRIPTS:
+    path = BASE / script
+    if not path.exists():
+        path = Path("/home/hmo/MoFin") / script
+    print(f"--- {script} ---", flush=True)
+    start = time.time()
+    try:
+        result = subprocess.run(
+            [sys.executable, str(path)],
+            capture_output=True, text=True, timeout=timeout
+        )
+        elapsed = time.time() - start
+        if result.returncode == 0:
+            print(f"OK ({elapsed:.0f}s)", flush=True)
+            if result.stdout.strip():
+                for line in result.stdout.strip().split("\n")[-3:]:
+                    print(f"  {line}", flush=True)
+        else:
+            print(f"FAIL ({elapsed:.0f}s): {result.stderr[:200]}", flush=True)
+    except subprocess.TimeoutExpired:
+        print(f"TIMEOUT ({timeout}s)", flush=True)
+    except Exception as e:
+        print(f"ERROR: {e}", flush=True)
+
+# ── 步骤4: 个股行情注入（唯一权威价格源）──
+# 所有持仓最新行情，注入到 LLM context
+# LLM 禁止自行调用原始API解析价格
+PRICE_SCRIPT = BASE / "stock_quote.py"
+if not PRICE_SCRIPT.exists():
+    PRICE_SCRIPT = Path("/home/hmo/MoFin/deploy/profile-scripts/stock_quote.py")
+if PRICE_SCRIPT.exists():
+    print("--- stock_quote.py ---", flush=True)
+    try:
+        result = subprocess.run(
+            [sys.executable, str(PRICE_SCRIPT), "--all-holdings"],
+            capture_output=True, text=True, timeout=30
+        )
+        if result.returncode == 0 and result.stdout.strip():
+            lines = [l for l in result.stdout.strip().split("\n") if l.strip()]
+            print(f"OK ({len(lines)}只持仓)", flush=True)
+            for line in lines[:50]:
+                print(f"  {line}", flush=True)
+        else:
+            print(f"WARN: stock_quote stderr={result.stderr[:100]}", flush=True)
+    except Exception as e:
+        print(f"WARN: stock_quote skipped ({e})", flush=True)
+else:
+    print("WARN: stock_quote.py not found", flush=True)
+
+print("采集链完成", flush=True)

archive/20260728-dedup/stale_detector.py

@@ -0,0 +1,397 @@
+#!/usr/bin/env python3
+"""stale_detector.py — 检查所有策略，标记价格偏离/过期的策略
+
+读取 holding_strategies + 自选策略的DB双源数据。
+可被 cron no_agent 模式调用：stdout 注入到后续 LLM 分析。
+
+输出格式：
+  [FLAG] [自选/持仓] 股票名(代码) 价XX | 买入A~B | 问题
+
+用法：
+  python3 stale_detector.py
+"""
+import json
+import sys
+import os
+from datetime import datetime, timezone
+sys.path.insert(0, '/home/hmo/MoFin')
+from mo_data import read_portfolio, read_decisions, read_watchlist, get_price, get_prices_batch
+
+
+def fetch_prices(codes):
+    """统一价格源：优先 stock_quote.py，腾讯API降级为兜底"""
+    if not codes:
+        return {}
+    # 尝试用 stock_quote.py 获取（脚本强制规范）
+    try:
+        import subprocess
+        script = None
+        for p in ["/home/hmo/MoFin/scripts/stock_quote.py", "/home/hmo/MoFin/stock_quote.py"]:
+            if os.path.exists(p):
+                script = p
+                break
+        if script:
+            result = subprocess.run(
+                [sys.executable, script] + [str(c) for c in codes],
+                capture_output=True, text=True, timeout=30
+            )
+            if result.returncode == 0 and result.stdout.strip():
+                results = {}
+                for line in result.stdout.strip().split("\n"):
+                    if not line.strip():
+                        continue
+                    try:
+                        item = json.loads(line)
+                        code = str(item.get("code", ""))
+                        price = item.get("price")
+                        change = item.get("change_pct", 0)
+                        if code and price is not None:
+                            results[code] = (float(price), float(change))
+                    except (json.JSONDecodeError, ValueError):
+                        continue
+                if results:
+                    return results
+    except Exception as e:
+        print(f"[STALE] stock_quote.py 回退: {e}", file=sys.stderr)
+
+    # 兜底：mo_data.get_prices_batch
+    try:
+        raw = get_prices_batch(codes)
+        if raw:
+            return {code: (p, chg) for code, (p, chg) in raw.items()}
+    except Exception as e:
+        print(f"FETCH_FAIL (fallback): {e}", file=sys.stderr)
+    return {}
+
+
+def main():
+    decisions_list = read_decisions()
+    if not isinstance(decisions_list, list):
+        decisions_list = decisions_list.get("decisions", []) if isinstance(decisions_list, dict) else []
+
+    # 只保留有买入区的条目，排除已关闭的（inactive/closed）
+    EXCLUDED_STATUSES = ("closed", "inactive")
+    to_check = [d for d in decisions_list if (d.get("entry_low") is not None or d.get("entry_high") is not None) and d.get("status") not in EXCLUDED_STATUSES]
+
+    # ----- 补充自选（从 holding_strategies 读取，watchlist_stocks 已废弃） -----
+    try:
+        import sqlite3
+        db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+        db.row_factory = sqlite3.Row
+        wl_rows = db.execute(
+            "SELECT code, name, entry_low, entry_high, stop_loss, take_profit, rr_ratio, timing_signal, action "
+            "FROM holding_strategies WHERE status='active' AND decision_type='自选策略' "
+            "AND entry_low IS NOT NULL AND entry_high IS NOT NULL"
+        ).fetchall()
+        db.close()
+        existing_codes = {d["code"] for d in to_check}
+        for row in wl_rows:
+            code = str(row["code"])
+            if code in existing_codes:
+                continue
+            entry_low = row["entry_low"]
+            entry_high = row["entry_high"]
+            if not entry_low or not entry_high or entry_low <= 0:
+                continue
+            action = row["action"] or ""
+            timing_signal = row["timing_signal"] or "买入"
+            wl_entry = {
+                "code": code,
+                "name": row["name"] or code,
+                "entry_low": entry_low,
+                "entry_high": entry_high,
+                "stop_loss": row["stop_loss"],
+                "type": "自选策略",
+                "action": action,
+                "timing_signal": timing_signal,
+            }
+            to_check.append(wl_entry)
+    except Exception as e:
+        print(f"[WATCHLIST_MERGE FAIL] {e}", file=sys.stderr)
+
+    if not to_check:
+        print("[SILENT] 无需要检查的策略")
+        return 0
+
+    # ----- 自选股买入区偏离自动重评 (从 holding_strategies 读，watchlist_stocks 已废弃) -----
+    try:
+        import subprocess, sqlite3
+        db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+        db.row_factory = sqlite3.Row
+        wl_stocks = db.execute(
+            "SELECT code, name, entry_low, entry_high "
+            "FROM holding_strategies WHERE status='active' AND decision_type='自选策略' "
+            "AND entry_low IS NOT NULL AND entry_high IS NOT NULL AND entry_low > 0"
+        ).fetchall()
+        db.close()
+        reassess_scripts = []
+        for ws in wl_stocks:
+            code, name, wl_el, wl_eh = ws
+            if not wl_el or not wl_el or wl_el <= 0:
+                continue
+            center = (wl_el + wl_eh) / 2
+            # 从 decisions 拿实时价
+            price_map = fetch_prices([code])
+            cur_price = price_map.get(code, (None, None))[0]
+            if not cur_price or cur_price <= 0:
+                continue
+            drift = (cur_price / center - 1) * 100
+            # 触发条件：价格偏离>15% 或 买入区明确错误（价格完全在区间外且偏离>50%）
+            price_outside = cur_price < wl_el or cur_price > wl_eh
+            if abs(drift) > 15 or (price_outside and abs(drift) > 50):
+                reassess_scripts.append(code)
+                print(f"[AUTO_REASSESS] {name}({code}) 价{cur_price:.2f}偏离买入区中心{center:.2f} {drift:+.0f}% → 触发重评")
+        if reassess_scripts:
+            # 调用 per_stock_reassess（每轮最多5只，防LLM慢导致整批超时；其余下轮继续）
+            reassess_path = None
+            for p in ['/home/hmo/MoFin/scripts/per_stock_reassess.py',
+                      '/home/hmo/.hermes/profiles/position-analyst/scripts/per_stock_reassess.py']:
+                if os.path.exists(p):
+                    reassess_path = p
+                    break
+            if reassess_path:
+                MAX_PER_RUN = 5
+                batch = reassess_scripts[:MAX_PER_RUN]
+                if len(reassess_scripts) > MAX_PER_RUN:
+                    print(f"[AUTO_REASSESS] 本轮限{MAX_PER_RUN}只，剩余{len(reassess_scripts)-MAX_PER_RUN}只下轮继续")
+                for code in batch:
+                    try:
+                        # LLM 重评冷启动 20-100s，deepseek-v4-pro 更慢 → 480s
+                        r = subprocess.run(['python3', reassess_path, code],
+                                         capture_output=True, text=True, timeout=480)
+                        out = r.stdout.strip()[:200] if r.stdout else ""
+                        err = r.stderr.strip()[:200] if r.stderr else ""
+                        print(f"  → {code}: exited={r.returncode} {out}")
+                    except subprocess.TimeoutExpired:
+                        print(f"  → {code}: 超时480s（LLM仍慢），下轮重试")
+    except Exception as e:
+        print(f"[AUTO_REASSESS FAIL] {e}")
+    # ----- 结束 自选股重评 -----
+    # 🔁 重评后重新从DB读取策略数据，刷新to_check
+    try:
+        decisions_list = read_decisions()
+        if not isinstance(decisions_list, list):
+            decisions_list = decisions_list.get("decisions", []) if isinstance(decisions_list, dict) else []
+        to_check = [d for d in decisions_list if (d.get("entry_low") is not None or d.get("entry_high") is not None) and d.get("status") not in EXCLUDED_STATUSES]
+        # 重新合并自选（从 holding_strategies 读）
+        db2 = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
+        db2.row_factory = sqlite3.Row
+        wl_rows2 = db2.execute(
+            "SELECT code, name, entry_low, entry_high, stop_loss, take_profit, rr_ratio, timing_signal, action "
+            "FROM holding_strategies WHERE status='active' AND decision_type='自选策略' "
+            "AND entry_low IS NOT NULL AND entry_high IS NOT NULL AND entry_low > 0"
+        ).fetchall()
+        db2.close()
+        existing_codes2 = {d["code"] for d in to_check}
+        for row in wl_rows2:
+            code = str(row["code"])
+            if code in existing_codes2:
+                continue
+            entry_low = row["entry_low"]
+            entry_high = row["entry_high"]
+            if not entry_low or not entry_high or entry_low <= 0:
+                continue
+            action = row["action"] or ""
+            timing_signal = row["timing_signal"] or "买入"
+            wl_entry = {
+                "code": code,
+                "name": row["name"] or code,
+                "entry_low": entry_low,
+                "entry_high": entry_high,
+                "stop_loss": row["stop_loss"],
+                "type": "自选策略",
+                "action": action,
+                "timing_signal": timing_signal,
+            }
+            to_check.append(wl_entry)
+    except Exception as e:
+        print(f"[RELOAD FAIL] {e}", file=sys.stderr)
+
+    # ----- 组合级监测：读取总仓位 + 弱势比例 -----
+    position_pct = 0
+    cash = 0
+    total_assets = 0
+    try:
+        pf = read_portfolio()
+        position_pct = pf.get("position_pct", 0)
+        cash = pf.get("cash", 0)
+        total_assets = pf.get("total_assets", 0)
+    except Exception:
+        pass
+    # 统计持仓策略中弱势/深套的比例
+    weak_count = 0
+    holding_count = 0
+    for d in decisions_list:
+        if d.get("type") == "持仓策略" and d.get("status") not in ("closed", "inactive"):
+            holding_count += 1
+            cat = d.get("stock_category", "")
+            if cat in ("弱势", "深套"):
+                weak_count += 1
+    weak_ratio = (weak_count / holding_count * 100) if holding_count > 0 else 0
+
+    prices = fetch_prices([d["code"] for d in to_check])
+    now = datetime.now(timezone.utc).astimezone()
+    found = 0
+
+    for d in to_check:
+        code = d["code"]
+        name = d.get("name", code)
+        el = d.get("entry_low")
+        eh = d.get("entry_high")
+        sl = d.get("stop_loss")
+        tp = d.get("take_profit")
+        ts = d.get("created_at") or d.get("timestamp") or d.get("updated_at", "")
+        is_wl = "自选" in (d.get("type", ""))
+
+        pi = prices.get(code)
+        if not pi:
+            continue
+        price, chg = pi
+        if price <= 0:
+            continue
+
+        issues, flags = [], []
+        tag = "[自选]" if is_wl else "[持仓]"
+
+        # -- 偏离 --
+        if is_wl and not issues and not flags:
+            # 自选在买入区上沿与20%之间（零标记漏洞）：标记为小幅偏离
+            if el and eh and price > eh:
+                flags.append("[WL_DRIFT]")
+                flags.append("[STRATEGY_STALE]")
+                issues.append(f"[STRATEGY_STALE] 价{price:.2f}超买入区上沿+{((price/eh)-1)*100:.1f}%，买入区需重评")
+        if is_wl and el and eh:
+            # 读取 timing_signal 判断策略有效性（timing_signal 字段优先，fallback to action）
+            current_str = d.get("current", "") or ""
+            timing_signal = d.get("timing_signal", "") or current_str
+            has_nonbuy_signal = any(kw in timing_signal for kw in [
+                "等企稳再入", "等企稳", "弱势持有", "观望",
+                "不建议买入", "谨慎买入",
+            ])
+
+            # 直接计算 R/R（不依赖文本匹配）
+            rr_invalid = False
+            if sl and sl > 0 and tp and tp > 0 and price > sl:
+                rr = (tp - price) / (price - sl)
+                if rr < 1.5:
+                    rr_invalid = True
+                # 也检查 tp 是否接近或低于成本（微盈/浮亏止盈）
+                cost = d.get("cost", 0)
+                if cost and cost > 0 and tp <= cost * 1.05:
+                    rr_invalid = True
+
+            strategy_deficient = has_nonbuy_signal or rr_invalid
+            # 对自选无止盈位的也标记（策略不完整）
+            if not tp or tp == 0:
+                strategy_deficient = True
+
+            if el <= price <= eh:
+                flags.append("[WL_IN]")
+                if strategy_deficient:
+                    flags.append("[STRATEGY_STALE]")
+                    issues.append(f"[STRATEGY_STALE] 价{price:.2f}在买入区{el}~{eh}但策略不完整（{'RR='+f'{rr:.2f}<1.5' if rr_invalid else '无止盈位' if not tp else '非买入信号'}），买入区需重评")
+                else:
+                    issues.append(f"[PUSH] 价{price:.2f}入买入区{el}~{eh}")
+            elif price > eh * 1.35:
+                flags.append("[WL_HIGH]")
+                flags.append("[STRATEGY_STALE]")
+                issues.append(f"[STRATEGY_STALE] 价{price:.2f}高出买入区+{((price/eh)-1)*100:.0f}%，买入区需重评")
+            elif price > eh * 1.20:
+                flags.append("[WL_DRIFT]")
+                flags.append("[STRATEGY_STALE]")
+                issues.append(f"[STRATEGY_STALE] 价{price:.2f}高出买入区+{((price/eh)-1)*100:.0f}%，买入区需重评")
+            elif price > eh:
+                flags.append("[WL_DRIFT]")
+                flags.append("[STRATEGY_STALE]")
+                issues.append(f"[STRATEGY_STALE] 价{price:.2f}超买入区上沿+{((price/eh)-1)*100:.1f}%，买入区需重评")
+        elif not is_wl and eh:
+            dp = (price / eh - 1) * 100
+            if dp > 35:
+                flags.append("[SEVERE]")
+                issues.append(f"偏离买入区上沿+{dp:.0f}%")
+            elif dp > 20:
+                flags.append("[DRIFT]")
+                issues.append(f"偏离买入区上沿+{dp:.0f}%")
+            elif dp > 10:
+                flags.append("[WARN]")
+                issues.append(f"偏离买入区上沿+{dp:.0f}%")
+            # 持仓在买入区内但 R/R 不达标
+            if el and sl and sl > 0 and tp and tp > 0 and price > sl:
+                if el <= price <= eh:
+                    rr = (tp - price) / (price - sl)
+                    if rr < 1.5:
+                        flags.append("[RR_WARN]")
+                        issues.append(f"买入区内RR仅{rr:.2f}<1.5，策略需重评")
+
+        # -- 距止损/止盈（仅持仓） --
+        if not is_wl:
+            if sl and sl > 0:
+                dsl = (price / sl - 1) * 100
+                if dsl < 5:
+                    # 成本基准校验：浮盈>5%时止损是利润保护，不是危险信号
+                    # (mirrors NEAR_TP cost_check logic at line 195-198)
+                    cost = d.get("cost")
+                    if cost and cost > 0 and price > cost * 1.05:
+                        flags.append("[PROFIT_PROTECT]")
+                        pnl = (price / cost - 1) * 100
+                        issues.append(f"距止损仅{dsl:.1f}%(利润保护,浮盈{pnl:.0f}%)")
+                    else:
+                        flags.append("[NEAR_SL]")
+                        issues.append(f"距止损仅{dsl:.1f}%")
+            if tp and tp > 0:
+                dtp = (tp / price - 1) * 100
+                if dtp < 5:
+                    # 成本基准校验：止盈标记只有在盈利≥5%时才有效
+                    cost_check = True
+                    cost = d.get("cost")
+                    if cost and cost > 0 and price < cost * 1.05:
+                        cost_check = False
+                    if cost_check:
+                        flags.append("[NEAR_TP]")
+                        issues.append(f"距止盈仅{dtp:.1f}%")
+
+        # -- 过期 --
+        stale_limit = 30 if is_wl else 14
+        if ts:
+            try:
+                ud = datetime.fromisoformat(ts)
+                if ud.tzinfo is None:
+                    ud = ud.replace(tzinfo=timezone.utc)
+                days = (now - ud).days
+                if days > stale_limit:
+                    flags.append("[STALE]")
+                    issues.append(f"{days}天未更新(>{stale_limit})")
+            except (ValueError, TypeError):
+                pass
+
+        if issues:
+            # 仅输出有明确操作信号的行：[PUSH]=推荐买入, [STRATEGY_STALE]=需重评
+            # 静默其他纯信息行（如仅"价XX高出/高于买入区"而无操作建议）
+            if any("[PUSH]" in i or "[STRATEGY_STALE]" in i for i in issues):
+                print(f"{' '.join(flags)} {tag} {name}({code}) 价{price:.2f}{chg} | 买入{el}~{eh} | {'; '.join(issues)}")
+                found += 1
+
+    if found == 0:
+        print("[SILENT] 所有策略正常")
+
+    # ----- 组合级警报 -----
+    portfolio_alerts = 0
+    if holding_count > 0:
+        if weak_ratio > 40:
+            print(f"\n[PORTFOLIO_WEAK] 组合弱势比例{weak_ratio:.0f}% ({weak_count}/{holding_count})！仓位{position_pct:.1f}% → 建议系统性减仓")
+            portfolio_alerts += 1
+        elif weak_ratio > 30:
+            print(f"\n[PORTFOLIO_WEAK_MILD] 组合弱势比例{weak_ratio:.0f}% ({weak_count}/{holding_count})，仓位{position_pct:.1f}%，关注")
+            portfolio_alerts += 1
+    if position_pct > 80 and holding_count > 0:
+        # 仓位过满提醒
+        print(f"[PORTFOLIO_FULL] 总仓位{position_pct:.1f}% > 80%，现金{cash:.0f}({cash/total_assets*100:.1f}%)")
+        portfolio_alerts += 1
+    if portfolio_alerts > 0:
+        found += portfolio_alerts
+
+    return found
+
+
+if __name__ == "__main__":
+    main()

mofin_collect.py

@@ -249,7 +249,7 @@ for script, timeout in SCRIPTS:
 # LLM 禁止自行调用原始API解析价格
 PRICE_SCRIPT = BASE / "stock_quote.py"
 if not PRICE_SCRIPT.exists():
-    PRICE_SCRIPT = Path("/home/hmo/MoFin/scripts/stock_quote.py")
+    PRICE_SCRIPT = Path("/home/hmo/MoFin/deploy/profile-scripts/stock_quote.py")
 if PRICE_SCRIPT.exists():
     print("--- stock_quote.py ---", flush=True)
     try: