diff --git a/scripts/stale_detector.py b/scripts/stale_detector.py index 41f18147..4f71089b 100644 --- a/scripts/stale_detector.py +++ b/scripts/stale_detector.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """stale_detector.py — 检查所有策略，标记价格偏离/过期的策略 -读取 holding_strategies + watchlist_stocks 的DB双源数据。 +读取 holding_strategies + 自选策略的DB双源数据。 可被 cron no_agent 模式调用：stdout 注入到后续 LLM 分析。 输出格式： @@ -73,14 +73,15 @@ def main(): 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] - # ----- 合并 DB watchlist_stocks 自选股 ----- + # ----- 补充自选（从 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, price, entry_low, entry_high, stop_loss, analysis_json " - "FROM watchlist_stocks WHERE is_active=1 AND entry_low IS NOT NULL AND entry_high IS NOT NULL" + "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} @@ -92,15 +93,8 @@ def main(): entry_high = row["entry_high"] if not entry_low or not entry_high or entry_low <= 0: continue - analysis = {} - aj = row["analysis_json"] - if aj: - try: - analysis = json.loads(aj) - except (json.JSONDecodeError, TypeError): - pass - action = analysis.get("action", "") if isinstance(analysis, dict) else "" - timing_signal = analysis.get("timing_signal", "买入") if isinstance(analysis, dict) else "买入" + action = row["action"] or "" + timing_signal = row["timing_signal"] or "买入" wl_entry = { "code": code, "name": row["name"] or code, @@ -119,19 +113,20 @@ def main(): print("[SILENT] 无需要检查的策略") return 0 - # ----- 自选股买入区偏离自动重评 (2026-07-07 fix: 不只标记, 直接触发) ----- + # ----- 自选股买入区偏离自动重评 (从 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, price, entry_low, entry_high, analysis_json " - "FROM watchlist_stocks WHERE is_active=1 AND entry_low IS NOT NULL" + "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_price, wl_el, wl_eh, wl_aj = ws + 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 @@ -168,12 +163,13 @@ def main(): 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] - # 重新合并watchlist_stocks + # 重新合并自选（从 holding_strategies 读） db2 = sqlite3.connect('/home/hmo/MoFin/data/mofin.db') db2.row_factory = sqlite3.Row wl_rows2 = db2.execute( - "SELECT code, name, price, entry_low, entry_high, stop_loss, analysis_json " - "FROM watchlist_stocks WHERE is_active=1 AND entry_low IS NOT NULL AND entry_high IS NOT NULL" + "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} @@ -185,13 +181,8 @@ def main(): entry_high = row["entry_high"] if not entry_low or not entry_high or entry_low <= 0: continue - analysis = {} - try: - analysis = json.loads(row["analysis_json"]) if row["analysis_json"] and row["analysis_json"] != "null" else {} - except (json.JSONDecodeError, TypeError): - pass - action = analysis.get("action", "") if isinstance(analysis, dict) else "" - timing_signal = analysis.get("timing_signal", "买入") if isinstance(analysis, dict) else "买入" + action = row["action"] or "" + timing_signal = row["timing_signal"] or "买入" wl_entry = { "code": code, "name": row["name"] or code, @@ -253,6 +244,12 @@ def main(): 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 "" @@ -293,6 +290,10 @@ def main(): 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: