From 069d4a603d3ac031d854ca580499d8af05a8cd3e Mon Sep 17 00:00:00 2001 From: hmo Date: Tue, 28 Jul 2026 09:08:13 +0800 Subject: [PATCH] fix: remove inactive concept, fix L3 noise --- deploy/profile-scripts/stale_detector.py | 794 +++++++++++------------ 1 file changed, 397 insertions(+), 397 deletions(-) diff --git a/deploy/profile-scripts/stale_detector.py b/deploy/profile-scripts/stale_detector.py index 30e1450c..81587482 100644 --- a/deploy/profile-scripts/stale_detector.py +++ b/deploy/profile-scripts/stale_detector.py @@ -1,397 +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/deploy/profile-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/deploy/profile-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() +#!/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/deploy/profile-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",) + 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/deploy/profile-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",): + 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()