diff --git a/deploy/profile-scripts/strategy-staleness-check.py b/deploy/profile-scripts/strategy-staleness-check.py index a77bf060..b0bcd392 100644 --- a/deploy/profile-scripts/strategy-staleness-check.py +++ b/deploy/profile-scripts/strategy-staleness-check.py @@ -1,269 +1,269 @@ -#!/usr/bin/env python3 -""" -策略时效性检查 - Strategy Staleness Checker v1 -============================================== -扫描所有活跃策略,检查三个维度: -1. 时间老化:超过14/21天未更新预警 -2. 价格偏离:当前价偏离买入区中心 >30% 预警 -3. 买入区失效:买入区完全脱离当前价格区间 - -输出两份:JSON报告(给系统) + 人类可读摘要(stdout for cron) -""" - -import json, sys, os, re, urllib.request, sqlite3 -sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) -from datetime import datetime -from mo_data import read_decisions, read_portfolio - -DB_PATH = '/home/hmo/web-dashboard/data/mofin.db' -OUTPUT_PATH = "/home/hmo/web-dashboard/data/strategy_staleness_report.json" - -WARN_DAYS = 14 # 超过14天未更新→警告 -CRITICAL_DAYS = 21 # 超过21天→严重警告 -DIVERGENCE_WARN = 30 # 偏离买入区>30%→警告 -DIVERGENCE_CRIT = 50 # 偏离>50%→严重 - -def get_price(code): - """从腾讯API获取当前价""" - try: - market = "sh" if code.startswith("6") else "sz" if code.startswith("0") or code.startswith("3") else "" - if code.startswith(("00", "30")) or code.startswith("68"): - market = "sh" if code.startswith("6") else "sz" - elif code.startswith(("01", "02", "03")): - market = "sz" - url = f"http://qt.gtimg.cn/q={market}{code}" - req = urllib.request.Request(url, headers={"User-Agent": "curl/7.81"}) - with urllib.request.urlopen(req, timeout=5) as resp: - raw = resp.read().decode("gbk") - parts = raw.split("~") - if len(parts) > 3: - price = float(parts[3]) if parts[3] else 0 - chg = float(parts[32]) if parts[32] else 0 - return price, chg if price > 0 else (None, None) - except: pass - return None, None - -def parse_buy_zone(current): - """从策略current字段提取买入区间最低和最高""" - if not current: - return None, None - m = re.search(r'买入.*?(\d+\.?\d*)\s*[~\-]\s*(\d+\.?\d*)', current) - if m: - return float(m.group(1)), float(m.group(2)) - return None, None - -def main(): - data = read_decisions() - - # Filter: exclude closed strategies - all_entries = data.get("decisions", []) - active = [e for e in all_entries if e.get("status", "") != "closed"] - flagged = [] - now = datetime.now() - - for entry in active: - code = entry.get("code", "") - name = entry.get("name", "") or code - entry_type = entry.get("type", "") - is_watchlist = "自选" in entry_type - - # --- Age tracking: prefer created_at, fallback to timestamp/updated_at --- - ts = entry.get("created_at") or entry.get("timestamp") or entry.get("updated_at") or "" - age = 0 - if ts: - try: - if "T" in ts: - dt = datetime.fromisoformat(ts) - else: - dt = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S") - age = (now - dt).days - except: - try: - # Try alternative format - dt = datetime.strptime(str(ts)[:19], "%Y-%m-%d %H:%M:%S") - age = (now - dt).days - except: - pass - - # --- Get price: prefer local data, API fallback --- - price = entry.get("price", None) - if not price or price == 0: - price, _ = get_price(code) - else: - price = float(price) if price else None - - # --- Get entry zone from structured fields --- - entry_low = entry.get("entry_low", None) - entry_high = entry.get("entry_high", None) - if (not entry_low) or (not entry_high): - # Fall back to parsing from action/current string - txt = entry.get("action", "") or entry.get("current", "") - entry_low, entry_high = parse_buy_zone(txt) - - # --- Build display info --- - action_text = entry.get("action", "") or entry.get("current", "") - last_update = entry.get("updated_at", "") or entry.get("timestamp", "") - - flags = [] - - # Time-based flagging (different thresholds for 持仓 vs 自选) - time_threshold = WARN_DAYS if not is_watchlist else WARN_DAYS * 2 # 自选给2倍时间 - critical_threshold = CRITICAL_DAYS if not is_watchlist else CRITICAL_DAYS * 2 - - if age >= critical_threshold and not is_watchlist: - flags.append(f"严重过期: {age}天未更新") - elif age >= time_threshold and is_watchlist: - flags.append(f"策略已{age}天未更新(自选)") - elif age >= time_threshold: - flags.append(f"策略已{age}天未更新") - - if price and entry_low and entry_high: - # -- STRATEGY_STALE check for watchlist stocks in buy zone -- - if is_watchlist and entry_low <= price <= entry_high: - timing_signal = entry.get("timing_signal", "") or "" - rr_ratio = entry.get("rr_ratio", 0) or 0 - sl = entry.get("stop_loss", 0) or 0 - tp = entry.get("take_profit", 0) or 0 - # 规则1: timing_signal 含"弱势持有"/"等企稳" → STRATEGY_STALE - if any(kw in timing_signal for kw in ["弱势持有", "等企稳"]): - flags.append("[STRATEGY_STALE] 信号不良(timing_signal含"+str([kw for kw in ["弱势持有", "等企稳"] if kw in timing_signal])+")") - # 规则2: RR<1.5 或无止盈 → STRATEGY_STALE - if tp == 0 or (rr_ratio > 0 and rr_ratio < 1.5): - flags.append("[STRATEGY_STALE] 盈亏比不足(RR="+str(rr_ratio)+")或无止盈") - - if is_watchlist: - # 自选股:检查价格是否进入了买入区 - if entry_low <= price <= entry_high: - flags.append(f"现价{price:.2f}在买入区{entry_low:.0f}~{entry_high:.0f}(是否可买需结合timing_signal判断)") - elif price > entry_high * 1.3: - flags.append(f"现价{price:.2f}远高于买入区{entry_low:.0f}~{entry_high:.0f},需重评") - elif price < entry_low * 0.8: - flags.append(f"现价{price:.2f}远低于买入区{entry_low:.0f}~{entry_high:.0f},买入区需下移") - else: - # 持仓股:检查价格偏离买入区中心 - zone_center = (entry_low + entry_high) / 2 - if zone_center > 0: - divergence = abs(price - zone_center) / zone_center * 100 - if divergence >= DIVERGENCE_CRIT: - flags.append(f"价格距买入区中心{divergence:.0f}% 已严重偏离") - elif divergence >= DIVERGENCE_WARN: - flags.append(f"价格距买入区中心{divergence:.0f}% 需关注") - - # Check if price entirely out of zone - if price > entry_high * 1.5: - flags.append(f"现价{price:.0f}远高于买入区{entry_low:.0f}~{entry_high:.0f}") - elif price < entry_low * 0.5: - flags.append(f"现价{price:.0f}远低于买入区{entry_low:.0f}~{entry_high:.0f}") - # Add type tag for clarity - type_tag = "自选" if is_watchlist else "持仓" - - if flags: - flagged.append({ - "code": code, - "name": name, - "price": round(price, 2) if price else None, - "flags": flags, - "age_days": age, - "last_update": last_update, - "entry_zone": f"{entry_low:.0f}~{entry_high:.0f}" if entry_low else "无买入区", - "current": action_text, - "updated_by": entry.get("source", "auto"), - "updated_reason": "自动生成", - "is_watchlist": is_watchlist, - }) - - # --- Portfolio-level analysis: count weak + deep_loss categories --- - holdings_count = len([e for e in active if "自选" not in e.get("type", "")]) - weak_count = len([e for e in active if e.get("stock_category") in ("弱势", "深套") and "自选" not in e.get("type", "")]) - all_weak_count = len([e for e in active if e.get("stock_category") in ("弱势", "深套")]) - weak_pct = round(weak_count / holdings_count * 100, 1) if holdings_count > 0 else 0 - all_weak_pct = round(all_weak_count / len(active) * 100, 1) if len(active) > 0 else 0 - - # Read portfolio for position% - position_pct = 0 - cash = 0 - try: - pdata = read_portfolio() - position_pct = pdata.get("position_pct", 0) - cash = pdata.get("cash", 0) - except: pass - - portfolio_flags = [] - if weak_pct >= 40: - portfolio_flags.append(f"[PORTFOLIO_WEAK] 组合中弱势+深套分类持仓占比{weak_pct}%>40%,建议系统性减仓") - elif weak_pct >= 30: - portfolio_flags.append(f"[PORTFOLIO_WEAK_MILD] 组合弱势占比{weak_pct}%,需关注") - if position_pct > 80: - portfolio_flags.append(f"[PORTFOLIO_FULL] 总仓位{position_pct}%(现金{cash:.0f}元),买入建议受限") - - # 写 TODO — 过期策略自动建修复任务 - stale_items = [s for s in flagged if any("[STRATEGY_STALE]" in f for f in s.get("flags", []))] - if stale_items: - conn = sqlite3.connect(DB_PATH) - for s in stale_items: - code = s["code"] - existing = conn.execute("SELECT id FROM todos WHERE title LIKE ? AND status IN ('pending','in_progress')", (f"%{code}%",)).fetchone() - if existing: - continue - conn.execute( - "INSERT INTO todos (title, description, status, priority, source, fix_action) VALUES (?,?,?,?,?,?)", - (f"[STRATEGY_STALE] {s['name']}({code}) 需重评", - f"策略过期: {'; '.join(s['flags'])} | 上次更新{s['age_days']}天前 | 区间{s['entry_zone']}", - "pending", "high", "staleness_check", - f"cd /home/hmo/MoFin && python3 scripts/per_stock_reassess.py {code}", - ) - ) - conn.commit() - conn.close() - print(f" TODO: 为 {len(stale_items)} 个过期策略创建修复任务") - - # Write report - os.makedirs(os.path.dirname(OUTPUT_PATH), exist_ok=True) - report = { - "checked_at": now.strftime("%Y-%m-%dT%H:%M:%S"), - "total_active": len(active), - "flagged_count": len(flagged), - "flagged": flagged, - "portfolio": { - "position_pct": position_pct, - "cash": round(cash, 2), - "weak_position_pct": weak_pct, - "all_weak_pct": all_weak_pct, - "signals": portfolio_flags - }, - "summary": f"扫描{len(active)}个策略,{len(flagged)}个需关注" - } - with open(OUTPUT_PATH, 'w') as f: - json.dump(report, f, indent=2, ensure_ascii=False) - - # Human-readable output - now_str = now.strftime("%m/%d %H:%M") - print(f"【策略重估检查】{now_str}") - print(f"活跃策略{len(active)}个 | 需关注{len(flagged)}个") - - # Portfolio-level signals first (if any) - for pf in portfolio_flags: - print(pf) - - if flagged: - print("") - for s in flagged: - price_str = f"¥{s['price']}" if s['price'] else "N/A" - ttag = "[自选]" if s.get('is_watchlist') else "" - print(f"{ttag}[{s['code']}] {s['name']} {price_str}") - print(f" 上次更新{s['age_days']}天前 | 区间{s['entry_zone']}") - for f in s['flags']: - print(f" ⚠ {f}") - print(f" 策略: {s['current']}") - print("") - - print(f"—END—{len(flagged)}个需关注 | 弱势持仓占比{weak_pct}% | 仓位{position_pct}%") - return len(flagged) - -if __name__ == "__main__": - try: - stale = main() - sys.exit(0) # Always exit 0 — flagged items are in JSON report + stdout; exit code 1 creates misleading "Script Error" alerts - except Exception as e: - print(f"ERROR: {e}") - sys.exit(2) +#!/usr/bin/env python3 +""" +策略时效性检查 - Strategy Staleness Checker v1 +============================================== +扫描所有活跃策略,检查三个维度: +1. 时间老化:超过14/21天未更新预警 +2. 价格偏离:当前价偏离买入区中心 >30% 预警 +3. 买入区失效:买入区完全脱离当前价格区间 + +输出两份:JSON报告(给系统) + 人类可读摘要(stdout for cron) +""" + +import json, sys, os, re, urllib.request, sqlite3 +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +from datetime import datetime +from mo_data import read_decisions, read_portfolio + +DB_PATH = '/home/hmo/web-dashboard/data/mofin.db' +OUTPUT_PATH = "/home/hmo/web-dashboard/data/strategy_staleness_report.json" + +WARN_DAYS = 14 # 超过14天未更新→警告 +CRITICAL_DAYS = 21 # 超过21天→严重警告 +DIVERGENCE_WARN = 30 # 偏离买入区>30%→警告 +DIVERGENCE_CRIT = 50 # 偏离>50%→严重 + +def get_price(code): + """从腾讯API获取当前价""" + try: + market = "sh" if code.startswith("6") else "sz" if code.startswith("0") or code.startswith("3") else "" + if code.startswith(("00", "30")) or code.startswith("68"): + market = "sh" if code.startswith("6") else "sz" + elif code.startswith(("01", "02", "03")): + market = "sz" + url = f"http://qt.gtimg.cn/q={market}{code}" + req = urllib.request.Request(url, headers={"User-Agent": "curl/7.81"}) + with urllib.request.urlopen(req, timeout=5) as resp: + raw = resp.read().decode("gbk") + parts = raw.split("~") + if len(parts) > 3: + price = float(parts[3]) if parts[3] else 0 + chg = float(parts[32]) if parts[32] else 0 + return price, chg if price > 0 else (None, None) + except: pass + return None, None + +def parse_buy_zone(current): + """从策略current字段提取买入区间最低和最高""" + if not current: + return None, None + m = re.search(r'买入.*?(\d+\.?\d*)\s*[~\-]\s*(\d+\.?\d*)', current) + if m: + return float(m.group(1)), float(m.group(2)) + return None, None + +def main(): + data = read_decisions() + + # Filter: exclude closed strategies + all_entries = data.get("decisions", []) + active = [e for e in all_entries if e.get("status", "") != "closed"] + flagged = [] + now = datetime.now() + + for entry in active: + code = entry.get("code", "") + name = entry.get("name", "") or code + entry_type = entry.get("type", "") + is_watchlist = "自选" in entry_type + + # --- Age tracking: prefer created_at, fallback to timestamp/updated_at --- + ts = entry.get("created_at") or entry.get("timestamp") or entry.get("updated_at") or "" + age = 0 + if ts: + try: + if "T" in ts: + dt = datetime.fromisoformat(ts) + else: + dt = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S") + age = (now - dt).days + except: + try: + # Try alternative format + dt = datetime.strptime(str(ts)[:19], "%Y-%m-%d %H:%M:%S") + age = (now - dt).days + except: + pass + + # --- Get price: prefer local data, API fallback --- + price = entry.get("price", None) + if not price or price == 0: + price, _ = get_price(code) + else: + price = float(price) if price else None + + # --- Get entry zone from structured fields --- + entry_low = entry.get("entry_low", None) + entry_high = entry.get("entry_high", None) + if (not entry_low) or (not entry_high): + # Fall back to parsing from action/current string + txt = entry.get("action", "") or entry.get("current", "") + entry_low, entry_high = parse_buy_zone(txt) + + # --- Build display info --- + action_text = entry.get("action", "") or entry.get("current", "") + last_update = entry.get("updated_at", "") or entry.get("timestamp", "") + + flags = [] + + # Time-based flagging (different thresholds for 持仓 vs 自选) + time_threshold = WARN_DAYS if not is_watchlist else WARN_DAYS * 2 # 自选给2倍时间 + critical_threshold = CRITICAL_DAYS if not is_watchlist else CRITICAL_DAYS * 2 + + if age >= critical_threshold and not is_watchlist: + flags.append(f"严重过期: {age}天未更新") + elif age >= time_threshold and is_watchlist: + flags.append(f"策略已{age}天未更新(自选)") + elif age >= time_threshold: + flags.append(f"策略已{age}天未更新") + + if price and entry_low and entry_high: + # -- STRATEGY_STALE check for watchlist stocks in buy zone -- + if is_watchlist and entry_low <= price <= entry_high: + timing_signal = entry.get("timing_signal", "") or "" + rr_ratio = entry.get("rr_ratio", 0) or 0 + sl = entry.get("stop_loss", 0) or 0 + tp = entry.get("take_profit", 0) or 0 + # 规则1: timing_signal 含"弱势持有"/"等企稳" → STRATEGY_STALE + if any(kw in timing_signal for kw in ["弱势持有", "等企稳"]): + flags.append("[STRATEGY_STALE] 信号不良(timing_signal含"+str([kw for kw in ["弱势持有", "等企稳"] if kw in timing_signal])+")") + # 规则2: RR<1.5 或无止盈 → STRATEGY_STALE + if tp == 0 or (rr_ratio > 0 and rr_ratio < 1.5): + flags.append("[STRATEGY_STALE] 盈亏比不足(RR="+str(rr_ratio)+")或无止盈") + + if is_watchlist: + # 自选股:检查价格是否进入了买入区 + if entry_low <= price <= entry_high: + flags.append(f"现价{price:.2f}在买入区{entry_low:.0f}~{entry_high:.0f}(是否可买需结合timing_signal判断)") + elif price > entry_high * 1.3: + flags.append(f"现价{price:.2f}远高于买入区{entry_low:.0f}~{entry_high:.0f},需重评") + elif price < entry_low * 0.8: + flags.append(f"现价{price:.2f}远低于买入区{entry_low:.0f}~{entry_high:.0f},买入区需下移") + else: + # 持仓股:检查价格偏离买入区中心 + zone_center = (entry_low + entry_high) / 2 + if zone_center > 0: + divergence = abs(price - zone_center) / zone_center * 100 + if divergence >= DIVERGENCE_CRIT: + flags.append(f"价格距买入区中心{divergence:.0f}% 已严重偏离") + elif divergence >= DIVERGENCE_WARN: + flags.append(f"价格距买入区中心{divergence:.0f}% 需关注") + + # Check if price entirely out of zone + if price > entry_high * 1.5: + flags.append(f"现价{price:.0f}远高于买入区{entry_low:.0f}~{entry_high:.0f}") + elif price < entry_low * 0.5: + flags.append(f"现价{price:.0f}远低于买入区{entry_low:.0f}~{entry_high:.0f}") + # Add type tag for clarity + type_tag = "自选" if is_watchlist else "持仓" + + if flags: + flagged.append({ + "code": code, + "name": name, + "price": round(price, 2) if price else None, + "flags": flags, + "age_days": age, + "last_update": last_update, + "entry_zone": f"{entry_low:.0f}~{entry_high:.0f}" if entry_low else "无买入区", + "current": action_text, + "updated_by": entry.get("source", "auto"), + "updated_reason": "自动生成", + "is_watchlist": is_watchlist, + }) + + # --- Portfolio-level analysis: count weak + deep_loss categories --- + holdings_count = len([e for e in active if "自选" not in e.get("type", "")]) + weak_count = len([e for e in active if e.get("stock_category") in ("弱势", "深套") and "自选" not in e.get("type", "")]) + all_weak_count = len([e for e in active if e.get("stock_category") in ("弱势", "深套")]) + weak_pct = round(weak_count / holdings_count * 100, 1) if holdings_count > 0 else 0 + all_weak_pct = round(all_weak_count / len(active) * 100, 1) if len(active) > 0 else 0 + + # Read portfolio for position% + position_pct = 0 + cash = 0 + try: + pdata = read_portfolio() + position_pct = pdata.get("position_pct", 0) + cash = pdata.get("cash", 0) + except: pass + + portfolio_flags = [] + if weak_pct >= 40: + portfolio_flags.append(f"[PORTFOLIO_WEAK] 组合中弱势+深套分类持仓占比{weak_pct}%>40%,建议系统性减仓") + elif weak_pct >= 30: + portfolio_flags.append(f"[PORTFOLIO_WEAK_MILD] 组合弱势占比{weak_pct}%,需关注") + if position_pct > 80: + portfolio_flags.append(f"[PORTFOLIO_FULL] 总仓位{position_pct}%(现金{cash:.0f}元),买入建议受限") + + # 写 TODO — 过期策略自动建修复任务 + stale_items = [s for s in flagged if any("[STRATEGY_STALE]" in f for f in s.get("flags", []))] + if stale_items: + conn = sqlite3.connect(DB_PATH) + for s in stale_items: + code = s["code"] + existing = conn.execute("SELECT id FROM todos WHERE title LIKE ? AND status IN ('pending','in_progress')", (f"%{code}%",)).fetchone() + if existing: + continue + conn.execute( + "INSERT INTO todos (title, description, status, priority, source, fix_action) VALUES (?,?,?,?,?,?)", + (f"[STRATEGY_STALE] {s['name']}({code}) 需重评", + f"策略过期: {'; '.join(s['flags'])} | 上次更新{s['age_days']}天前 | 区间{s['entry_zone']}", + "pending", "high", "staleness_check", + f"cd /home/hmo/MoFin && python3 deploy/profile-scripts/per_stock_reassess.py {code}", + ) + ) + conn.commit() + conn.close() + print(f" TODO: 为 {len(stale_items)} 个过期策略创建修复任务") + + # Write report + os.makedirs(os.path.dirname(OUTPUT_PATH), exist_ok=True) + report = { + "checked_at": now.strftime("%Y-%m-%dT%H:%M:%S"), + "total_active": len(active), + "flagged_count": len(flagged), + "flagged": flagged, + "portfolio": { + "position_pct": position_pct, + "cash": round(cash, 2), + "weak_position_pct": weak_pct, + "all_weak_pct": all_weak_pct, + "signals": portfolio_flags + }, + "summary": f"扫描{len(active)}个策略,{len(flagged)}个需关注" + } + with open(OUTPUT_PATH, 'w') as f: + json.dump(report, f, indent=2, ensure_ascii=False) + + # Human-readable output + now_str = now.strftime("%m/%d %H:%M") + print(f"【策略重估检查】{now_str}") + print(f"活跃策略{len(active)}个 | 需关注{len(flagged)}个") + + # Portfolio-level signals first (if any) + for pf in portfolio_flags: + print(pf) + + if flagged: + print("") + for s in flagged: + price_str = f"¥{s['price']}" if s['price'] else "N/A" + ttag = "[自选]" if s.get('is_watchlist') else "" + print(f"{ttag}[{s['code']}] {s['name']} {price_str}") + print(f" 上次更新{s['age_days']}天前 | 区间{s['entry_zone']}") + for f in s['flags']: + print(f" ⚠ {f}") + print(f" 策略: {s['current']}") + print("") + + print(f"—END—{len(flagged)}个需关注 | 弱势持仓占比{weak_pct}% | 仓位{position_pct}%") + return len(flagged) + +if __name__ == "__main__": + try: + stale = main() + sys.exit(0) # Always exit 0 — flagged items are in JSON report + stdout; exit code 1 creates misleading "Script Error" alerts + except Exception as e: + print(f"ERROR: {e}") + sys.exit(2)