276 lines
12 KiB
Python
276 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
策略时效性检查 - Strategy Staleness Checker v1
|
||
==============================================
|
||
扫描所有活跃策略，检查三个维度：
|
||
1. 时间老化：超过14/21天未更新预警
|
||
2. 价格偏离：当前价偏离买入区中心 >30% 预警
|
||
3. 买入区失效：买入区完全脱离当前价格区间
|
||
|
||
输出两份：JSON报告(给系统) + 人类可读摘要(stdout for cron)
|
||
"""
|
||
|
||
import json, sys, os, re, sqlite3
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||
from datetime import datetime
|
||
from mo_data import read_decisions, read_portfolio
|
||
|
||
# ── 消息通道统一路由(broadcast/xmpp by delivery) ──
|
||
try:
|
||
from messenger import install_stdio_hook as _msh
|
||
_msh()
|
||
except Exception:
|
||
pass
|
||
|
||
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):
|
||
"""从 live_prices 表获取当前价（2026-08-26 分层铁律：消费层不直连腾讯API）"""
|
||
try:
|
||
c = str(code).strip().split("_")[0]
|
||
if c.lower().startswith("hk"):
|
||
c = c[2:]
|
||
conn = sqlite3.connect(DB_PATH, timeout=5)
|
||
row = conn.execute(
|
||
"SELECT price, change_pct FROM live_prices WHERE code=?", (c,)
|
||
).fetchone()
|
||
conn.close()
|
||
if row and row[0]:
|
||
price = float(row[0])
|
||
chg = float(row[1] or 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)
|
||
conn.execute("PRAGMA busy_timeout=30000") # 2026-08-28 防并发写锁(DB损坏教训)
|
||
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)
|