From 835314fbb0ea9e5d4eff2210c541291955026e9e Mon Sep 17 00:00:00 2001 From: hmo Date: Tue, 28 Jul 2026 08:47:39 +0800 Subject: [PATCH] cleanup: remove stale scripts/ copies (authoritative in deploy/profile-scripts/) --- scripts/candidate_filter.py | 354 -------------------------------- scripts/stale_detector.py | 397 ------------------------------------ 2 files changed, 751 deletions(-) delete mode 100644 scripts/candidate_filter.py delete mode 100644 scripts/stale_detector.py diff --git a/scripts/candidate_filter.py b/scripts/candidate_filter.py deleted file mode 100644 index 4a48c068..00000000 --- a/scripts/candidate_filter.py +++ /dev/null @@ -1,354 +0,0 @@ -#!/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() diff --git a/scripts/stale_detector.py b/scripts/stale_detector.py deleted file mode 100644 index fc805b83..00000000 --- a/scripts/stale_detector.py +++ /dev/null @@ -1,397 +0,0 @@ -#!/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()