405 lines
18 KiB
Python
405 lines
18 KiB
Python
#!/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
|
||
|
||
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
||
try:
|
||
from messenger import install_stdio_hook as _msh
|
||
_msh()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
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()
|