hmo
d5b8bec897
User directive: no JSON, retire it fully, fix all related code. Changes: - price_monitor.py: record_event writes DB only; removed EVENTS_PATH/ load_events/save_events entirely - strategy_feedback.py: price events read from DB only (removed JSON fallback) - system_health_check.py: removed price_events.json from file-check list, DB-only event stats (was showing 0/0 due to wrong-DB resolution) - mo_config.py: removed dead price_events_path property (no callers) - mofin_health.py: price_events freshness reads DB table (authoritative now) - mofin_db.py: DATA_DIR/DB_PATH now ABSOLUTE (/home/hmo/MoFin/data) — was relative __file__.parent, so each hardlinked copy of mofin_db.py resolved to a DIFFERENT database (canonical vs web-dashboard vs profile-local third DB with 0 rows of everything except market_snapshots). This fragmentation was the real cause of health checks reading empty tables. - Unified all 4 mofin_db copies (root/scripts/deploy/profile) via hardlink - price_events.json archived to trashbox (fully backfilled: 6353 rows in DB) Verified: - record_event lands in DB only, JSON not recreated - system_health_check: 历史事件 6353 / 今日事件 2965 (was 0/0) - strategy_feedback + price_monitor full runs clean
733 lines
30 KiB
Python
733 lines
30 KiB
Python
#!/usr/bin/env python3
|
||
"""price_monitor.py — 高频价格监控脚本（批量版）
|
||
规则：进入区间报一次，离开区间报一次，中间不重复。
|
||
每次运行时一次性刷新所有持仓+自选股的实时价。
|
||
"""
|
||
import urllib.request
|
||
import os, sys, time, json
|
||
import sqlite3
|
||
from datetime import datetime
|
||
|
||
from mo_data import read_decisions
|
||
|
||
BREACH_PATH = "/home/hmo/.hermes/zone_breach.json"
|
||
STATE_PATH = "/home/hmo/.hermes/price_trigger_state.json"
|
||
|
||
# DB 模块（同步实时价到 mofin.db）
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
try:
|
||
from mofin_db import get_conn, DB_PATH
|
||
from mo_models import calc_total_mv, calc_total_assets
|
||
HAS_DB = True
|
||
except ImportError:
|
||
HAS_DB = False
|
||
|
||
# 策略重评依赖（技术面驱动，非机械百分比）
|
||
sys.path.insert(0, "/home/hmo/web-dashboard")
|
||
try:
|
||
from strategy_lifecycle import reassess_strategy, reassess_with_context
|
||
HAS_REASSESS = True
|
||
except ImportError:
|
||
HAS_REASSESS = False
|
||
|
||
UA = "Mozilla/5.0"
|
||
|
||
# ── XMPP推送 ──────────────────────────────────────────────────────────
|
||
XMPP_USER = "hmo@yoin.fun"
|
||
XMPP_BRIDGE = "http://127.0.0.1:5805/"
|
||
|
||
def push_to_xmpp(text):
|
||
"""通过知微 HTTP bridge 推送到Dad私信"""
|
||
if not text.strip():
|
||
return
|
||
try:
|
||
payload = json.dumps({
|
||
"to": XMPP_USER,
|
||
"body": text.strip(),
|
||
"type": "chat",
|
||
}).encode("utf-8")
|
||
req = urllib.request.Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"})
|
||
urllib.request.urlopen(req, timeout=5)
|
||
except Exception as e:
|
||
print(f"[XMPP推送失败] {e}", file=sys.stderr)
|
||
|
||
# ── 批量拉取价格 ──────────────────────────────────────────────────────────
|
||
|
||
def fetch_all_prices(codes):
|
||
"""腾讯批量行情API：一次请求拉取所有股票（A股+港股）
|
||
A股：sh600110 / sz000001
|
||
港股：hk00700
|
||
返回 {code: (price, change, change_pct)}
|
||
"""
|
||
if not codes:
|
||
return {}
|
||
|
||
# 构建批量查询串
|
||
symbols = []
|
||
code_map = {} # symbol -> original_code
|
||
for code in codes:
|
||
code_s = str(code).strip()
|
||
if len(code_s) == 6:
|
||
# A股：沪市以5/6/9开头，深市以0/3开头
|
||
if code_s.startswith(('5', '6', '9')):
|
||
sym = f"sh{code_s}"
|
||
else:
|
||
sym = f"sz{code_s}"
|
||
else:
|
||
sym = f"hk{code_s}"
|
||
symbols.append(sym)
|
||
code_map[sym] = code_s
|
||
|
||
url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
|
||
try:
|
||
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
||
with urllib.request.urlopen(req, timeout=10) as r:
|
||
text = r.read().decode("gbk")
|
||
except Exception as e:
|
||
print(f"⚠️ 批量拉取失败: {e}", file=sys.stderr)
|
||
return {}
|
||
|
||
results = {}
|
||
for line in text.strip().split("\n"):
|
||
line = line.strip()
|
||
if not line or "=" not in line:
|
||
continue
|
||
try:
|
||
# 格式: v_sh600110="1~诺德股份~600110~11.84~11.90~..."
|
||
raw_value = line.split("=", 1)[1].strip().strip('"').strip(";")
|
||
fields = raw_value.split("~")
|
||
if len(fields) < 6:
|
||
continue
|
||
sym = line.split("=", 1)[0].strip().lstrip("v_")
|
||
orig_code = code_map.get(sym)
|
||
if not orig_code:
|
||
continue
|
||
price = float(fields[3]) if fields[3] else 0
|
||
prev_close = float(fields[4]) if fields[4] else 0
|
||
change = price - prev_close if prev_close > 0 else 0
|
||
change_pct = fields[32] if len(fields) > 32 and fields[32] else "0"
|
||
results[orig_code] = (price, change, change_pct)
|
||
except (ValueError, IndexError):
|
||
continue
|
||
|
||
return results
|
||
|
||
|
||
def refresh_data_prices():
|
||
"""一次性刷新所有持仓+自选股的实时价（完全DB版，不写JSON）"""
|
||
all_codes = set()
|
||
|
||
# 从DB读所有需要拉取价格的代码
|
||
try:
|
||
conn = get_conn()
|
||
for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"):
|
||
all_codes.add(r['code'])
|
||
for r in conn.execute("SELECT code FROM watchlist_stocks"):
|
||
all_codes.add(r['code'])
|
||
for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"):
|
||
all_codes.add(r['code'])
|
||
conn.close()
|
||
except Exception as e:
|
||
print(f"⚠️ 从DB读代码失败: {e}", file=sys.stderr)
|
||
return 0
|
||
|
||
if not all_codes:
|
||
return 0
|
||
|
||
# 一次性批量拉取
|
||
prices = fetch_all_prices(list(all_codes))
|
||
updated = len(prices)
|
||
|
||
# === 弹性同步实时价到 mofin.db ===
|
||
# 防死锁策略（经2026-07-14 WAL死锁复盘改进）：
|
||
# ① 启动时 checkpoint WAL（清理残留事务）
|
||
# ② 统一 BEGIN IMMEDIATE 包裹整个写操作
|
||
# ③ 5次重试 + 指数退避: 1s → 2s → 4s → 8s → 16s（共~31s）
|
||
# ④ get_conn() 的 busy_timeout=30000 保证等待上限
|
||
# ⑤ 每个写操作检查返回值，任一失败立即 rollback + 重试
|
||
# ⑥ try/finally 确保连接始终释放
|
||
if HAS_DB and prices:
|
||
# 先checkpoint一次，清理上次被kill残留的WAL
|
||
try:
|
||
c = get_conn()
|
||
c.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||
c.close()
|
||
except Exception:
|
||
pass
|
||
|
||
max_tries = 5
|
||
conn = None
|
||
for db_attempt in range(max_tries):
|
||
try:
|
||
conn = get_conn()
|
||
# BEGIN IMMEDIATE 立即获取写锁——失败则等 busy_timeout(30s)
|
||
conn.execute("BEGIN IMMEDIATE")
|
||
|
||
# ── 构建 holdings 更新数据 ──
|
||
db_holdings = []
|
||
for r in conn.execute("SELECT * FROM holdings WHERE is_active=1"):
|
||
h = dict(r)
|
||
code = str(h.get('code', ''))
|
||
if code in prices:
|
||
price_val, _, change_pct = prices[code]
|
||
if price_val > 0:
|
||
h['price'] = round(price_val, 2)
|
||
h['change_pct'] = float(change_pct) if change_pct else 0
|
||
db_holdings.append(h)
|
||
|
||
# ── 写 holdings 表 ──
|
||
for h in db_holdings:
|
||
currency = str(h.get('currency', 'CNY')).upper()
|
||
if currency not in ('CNY', 'HKD'):
|
||
raise ValueError(f"非法币种: {currency}")
|
||
conn.execute("""
|
||
INSERT INTO holdings (code, name, shares, cost, price, market_value,
|
||
change_pct, currency, position_pct, added_at, is_active)
|
||
VALUES (?,?,?,?,?,?,?,?,?,datetime('now','localtime'),1)
|
||
ON CONFLICT(code) DO UPDATE SET
|
||
name=excluded.name, shares=excluded.shares, cost=excluded.cost,
|
||
price=excluded.price, market_value=excluded.market_value,
|
||
change_pct=excluded.change_pct, currency=excluded.currency,
|
||
position_pct=excluded.position_pct
|
||
""", (
|
||
h.get('code'), h.get('name'), h.get('shares', 0),
|
||
h.get('cost'), h.get('price'),
|
||
h.get('market_value'), h.get('change_pct'),
|
||
h.get('currency', 'CNY'), h.get('position_pct'),
|
||
))
|
||
|
||
# ── 写 portfolio_summary ──
|
||
mv = calc_total_mv(db_holdings)
|
||
existing = conn.execute(
|
||
'SELECT cash, frozen_cash FROM portfolio_summary WHERE id=1'
|
||
).fetchone()
|
||
db_cash = existing['cash'] if existing else 0.0
|
||
db_frozen = existing['frozen_cash'] if existing else 0.0
|
||
assets = calc_total_assets({'holdings': db_holdings, 'cash': db_cash, 'frozen_cash': db_frozen})
|
||
position_pct = round(mv / assets * 100, 2) if assets > 0 else 0
|
||
conn.execute("""
|
||
INSERT INTO portfolio_summary (id, total_assets, total_mv, stock_value,
|
||
cash, frozen_cash, position_pct, total_pnl, currency, updated_at)
|
||
VALUES (1,?,?,?,?,?,?,?,?,datetime('now','localtime'))
|
||
ON CONFLICT(id) DO UPDATE SET
|
||
total_assets=excluded.total_assets, total_mv=excluded.total_mv,
|
||
stock_value=excluded.stock_value, cash=excluded.cash,
|
||
frozen_cash=excluded.frozen_cash, position_pct=excluded.position_pct,
|
||
total_pnl=excluded.total_pnl, currency=excluded.currency,
|
||
updated_at=datetime('now','localtime')
|
||
""", (
|
||
assets, mv, mv, db_cash, db_frozen,
|
||
position_pct, 0, 'CNY',
|
||
))
|
||
|
||
# ── 写 live_prices ──
|
||
for h in db_holdings:
|
||
code = h.get('code', '')
|
||
if code:
|
||
p = h.get('price', 0)
|
||
cp = h.get('change_pct', 0)
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) "
|
||
"VALUES (?,?,?,datetime('now','localtime'))",
|
||
(code, p, cp)
|
||
)
|
||
# 补充策略股/自选股的价格（不在holdings中的）
|
||
for code, pdata in prices.items():
|
||
if code not in {h.get('code') for h in db_holdings}:
|
||
price_val = pdata[0] if isinstance(pdata, (list, tuple)) else pdata.get('price', 0)
|
||
cp_val = pdata[1] if isinstance(pdata, (list, tuple)) else pdata.get('change_pct', 0)
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) "
|
||
"VALUES (?,?,?,datetime('now','localtime'))",
|
||
(code, price_val, cp_val)
|
||
)
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
conn = None
|
||
if db_attempt > 0:
|
||
print(f"DB同步成功（第{db_attempt+1}次重试）")
|
||
break # success
|
||
|
||
except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
|
||
if conn:
|
||
try: conn.rollback()
|
||
except Exception: pass
|
||
try: conn.close()
|
||
except Exception: pass
|
||
conn = None
|
||
err_str = str(e)
|
||
if "locked" in err_str or "cannot commit" in err_str or "busy" in err_str:
|
||
if db_attempt < max_tries - 1:
|
||
wait = 2 ** db_attempt # 1, 2, 4, 8, 16
|
||
print(f"⏳ DB锁(尝试{db_attempt+1}/{max_tries}): {e} → {wait}s后重试", file=sys.stderr)
|
||
time.sleep(wait)
|
||
else:
|
||
print(f"❌ DB锁(重试{max_tries}次耗尽): {e}", file=sys.stderr)
|
||
else:
|
||
print(f"❌ DB错误: {e}", file=sys.stderr)
|
||
break
|
||
except Exception as e:
|
||
if conn:
|
||
try: conn.rollback()
|
||
except Exception: pass
|
||
try: conn.close()
|
||
except Exception: pass
|
||
conn = None
|
||
print(f"⚠️ DB同步异常: {e}", file=sys.stderr)
|
||
break
|
||
else:
|
||
# for-else: loop exhausted without break
|
||
print("❌ DB同步失败（所有重试耗尽）", file=sys.stderr)
|
||
# 尝试紧急 WAL checkpoint（释放死锁）
|
||
try:
|
||
c = sqlite3.connect(str(DB_PATH), timeout=1)
|
||
c.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||
c.close()
|
||
print(" ↪ 紧急WAL checkpoint完成", file=sys.stderr)
|
||
except Exception as we:
|
||
print(f" ↪ WAL checkpoint也失败: {we}", file=sys.stderr)
|
||
|
||
return updated
|
||
|
||
|
||
# ── 区间偏离检测 ──────────────────────────────────────────────────────────
|
||
|
||
def load_state():
|
||
try:
|
||
with open(STATE_PATH) as f:
|
||
return json.load(f)
|
||
except:
|
||
return {}
|
||
|
||
def save_state(state):
|
||
os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
|
||
with open(STATE_PATH, 'w') as f:
|
||
json.dump(state, f, ensure_ascii=False, indent=2)
|
||
|
||
def load_breaches():
|
||
try:
|
||
with open(BREACH_PATH) as f:
|
||
return json.load(f)
|
||
except:
|
||
return {}
|
||
|
||
def save_breaches(data):
|
||
os.makedirs(os.path.dirname(BREACH_PATH), exist_ok=True)
|
||
with open(BREACH_PATH, 'w') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def record_event(code, name, event_type, price, trigger_value, event_label=""):
|
||
"""记录一次价格触发事件到 DB price_events 表（唯一权威存储，JSON 已退役）。
|
||
|
||
price_events.code 有 FK -> stocks(code)，未注册的股票（新候选/港股）
|
||
先注册再写事件，否则 FK 失败事件丢失。
|
||
"""
|
||
now = datetime.now().isoformat()
|
||
|
||
if HAS_DB:
|
||
try:
|
||
from mofin_db import get_conn, write_price_event
|
||
_c = get_conn()
|
||
_exch, _typ = ("HK", "H") if len(str(code)) == 5 else (("SH", "A") if str(code).startswith(("6", "9")) else ("SZ", "A"))
|
||
_c.execute("INSERT OR IGNORE INTO stocks (code, name, exchange, type, updated_at) VALUES (?,?,?,?,?)",
|
||
(str(code), name or str(code), _exch, _typ, now))
|
||
_c.commit()
|
||
write_price_event(_c, code=code, name=name, event_type=event_type,
|
||
price=round(price, 2), trigger_value=str(trigger_value),
|
||
event_label=event_label)
|
||
_c.close()
|
||
except Exception as e:
|
||
print(f"[price_events DB写入失败] {e}", file=sys.stderr)
|
||
|
||
|
||
def get_trigger_zones(trigger):
|
||
"""返回该trigger所有可监控的区间列表，跳过已执行的batch"""
|
||
zones = []
|
||
for key, label in [
|
||
("entry_zone", "加仓区间"),
|
||
("batch1_price", "试仓区间"),
|
||
("batch2_price", "加仓区间"),
|
||
("take_profit_zone", "止盈区间"),
|
||
("watch_low", "关注区间"),
|
||
("watch_high", "减仓区间"),
|
||
("watch_break", "止损区间")
|
||
]:
|
||
status_key = key.replace("_price", "_status")
|
||
if status_key in trigger and trigger[status_key] == "executed":
|
||
continue
|
||
val = trigger.get(key, "")
|
||
if val and "~" in val:
|
||
try:
|
||
parts = val.split("~")
|
||
lo, hi = float(parts[0]), float(parts[1])
|
||
zones.append((key, label, lo, hi))
|
||
except:
|
||
pass
|
||
sl = trigger.get("stop_loss", "")
|
||
if sl:
|
||
try:
|
||
sl_price = float(sl) if isinstance(sl, (int, float)) else float(sl)
|
||
zones.append(("stop_loss", "止损", 0, sl_price))
|
||
except:
|
||
pass
|
||
return zones
|
||
|
||
|
||
def _cleanup_lock():
|
||
"""清理进程锁文件"""
|
||
try:
|
||
os.remove("/tmp/price_monitor.lock")
|
||
except Exception:
|
||
pass
|
||
|
||
def _handle_sigterm(signum, frame):
|
||
"""收到SIGTERM时清理锁文件后退出"""
|
||
_cleanup_lock()
|
||
sys.exit(0)
|
||
|
||
def run_once(round_label=""):
|
||
"""执行一轮完整的监控流程"""
|
||
import os, signal # 必须在开头import，否则os变量会被后面的局部import绑定覆盖
|
||
signal.signal(signal.SIGTERM, _handle_sigterm)
|
||
os.nice(10) # 降低优先级，避免与DB其他写操作抢占
|
||
# ── 进程锁：同一时间只跑一个实例 ──
|
||
_lk = "/tmp/price_monitor.lock"
|
||
_pid = None
|
||
try:
|
||
with open(_lk) as _f:
|
||
_pid = int(_f.read().strip())
|
||
os.kill(_pid, 0)
|
||
print(f"[LOCK] 已有实例(PID {_pid})在运行，跳过本轮", file=sys.stderr, flush=True)
|
||
return
|
||
except (FileNotFoundError, ProcessLookupError, ValueError):
|
||
pass
|
||
with open(_lk, "w") as _f:
|
||
_f.write(str(os.getpid()))
|
||
|
||
label = f" [{round_label}]" if round_label else ""
|
||
start = time.time()
|
||
TIME_BUDGET = 90 # 预留30s给输出和清理，90s内必须完成核心逻辑
|
||
|
||
# === 第一步：一次性刷新所有价格 ===
|
||
refreshed = refresh_data_prices()
|
||
|
||
# === 第二步：检查触发条件 ===
|
||
try:
|
||
dec = read_decisions()
|
||
except:
|
||
print(f"❌{label} 无法读取decisions(DB)", file=sys.stderr)
|
||
return
|
||
|
||
active = [d for d in dec.get("decisions", []) if d.get("status") == "active"]
|
||
state = load_state()
|
||
outputs = []
|
||
state_updated = False
|
||
# 时间冷却：同股同区间30分钟内不重复推
|
||
_push_cooldown = {}
|
||
_cooldown_file = "/home/hmo/.hermes/.price_push_cooldown.json"
|
||
try:
|
||
import os
|
||
if os.path.exists(_cooldown_file):
|
||
with open(_cooldown_file) as _f:
|
||
_push_cooldown = json.load(_f)
|
||
except Exception:
|
||
_push_cooldown = {}
|
||
|
||
def _can_push(code, zone_key):
|
||
now = time.time()
|
||
key = f"{code}_{zone_key}"
|
||
last = _push_cooldown.get(key, 0)
|
||
if now - last < 1800: # 30分钟
|
||
return False
|
||
_push_cooldown[key] = now
|
||
# 持久化写入
|
||
try:
|
||
with open(_cooldown_file, "w") as _f:
|
||
json.dump(_push_cooldown, _f)
|
||
except Exception:
|
||
pass
|
||
return True
|
||
|
||
# 收集所有需要检查的代码
|
||
check_codes = set()
|
||
for d in active:
|
||
trig = d.get("trigger", {})
|
||
if trig:
|
||
check_codes.add(d["code"])
|
||
|
||
# 批量拉取这些股票的价格
|
||
prices = fetch_all_prices(list(check_codes))
|
||
|
||
for d in active:
|
||
code = d["code"]
|
||
trig = d.get("trigger", {})
|
||
if not trig:
|
||
continue
|
||
|
||
zones = get_trigger_zones(trig)
|
||
if not zones:
|
||
continue
|
||
|
||
price_info = prices.get(code)
|
||
if not price_info:
|
||
continue
|
||
price, _, _ = price_info
|
||
if price == 0:
|
||
continue
|
||
|
||
name = d.get("name", code)
|
||
if code not in state:
|
||
state[code] = {}
|
||
|
||
# 时间预算检查：如果超时，跳过重评只做状态记录
|
||
_budget_low = (time.time() - start) > TIME_BUDGET
|
||
|
||
for key, label, lo, hi in zones:
|
||
in_zone = lo <= price <= hi
|
||
prev_in_zone = state[code].get(key, None)
|
||
|
||
if in_zone and prev_in_zone != True:
|
||
if key == "stop_loss":
|
||
outputs.append(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！")
|
||
record_event(code, name, "stop_loss", price, str(hi))
|
||
# 止损触发 → 立即重评并推送给Dad（时间不够则直接推原始告警）
|
||
if _budget_low:
|
||
outputs.append(f" 📨 止损触发(超时跳过重评)→已推送Dad")
|
||
if _can_push(code, "stop_loss"):
|
||
push_to_xmpp(f"⚠️ {name}({code}) {price} → 跌破止损{hi}！")
|
||
else:
|
||
try:
|
||
cost = d.get("cost", 0) or 0
|
||
shares = d.get("shares", 0) or 0
|
||
current_action = d.get("action", "")
|
||
result = reassess_with_context(code, name, price, cost, shares, current_action)
|
||
if result:
|
||
timing_signal = result.get("timing_signal", "")
|
||
action = result.get("action", "")
|
||
if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"):
|
||
buy_lo = d.get("entry_low", 0)
|
||
buy_hi = d.get("entry_high", 0)
|
||
rr = result.get("rr_ratio", 0)
|
||
if _can_push(code, "stop_loss"):
|
||
msg = f"🔔 {name}({code}) 价{price}→触发操作区间{max(buy_lo,0):.2f}~{buy_hi:.2f}，已触发重评|RR={rr}"
|
||
push_to_xmpp(msg)
|
||
outputs.append(f" 📨 止损重评→已推送Dad: {action}")
|
||
except Exception as e:
|
||
outputs.append(f" ⚠️ 止损重评失败: {e}")
|
||
else:
|
||
extra = ""
|
||
if "_price" in key:
|
||
batch_shares = trig.get(key.replace("_price", "_shares"), "")
|
||
action = trig.get(key.replace("_price", "_action"), "")
|
||
if batch_shares:
|
||
extra = f" {action}{batch_shares}股" if action else f" {batch_shares}股"
|
||
elif key in ("take_profit_zone",):
|
||
act = trig.get("take_profit_action", "")
|
||
if act:
|
||
extra = f"（{act}）"
|
||
outputs.append(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}{extra}")
|
||
record_event(code, name, "entry_zone", price, f"{lo}~{hi}", label)
|
||
# 进入区间 → 立即重评并推送给Dad（时间不够则跳过重评直接推原始告警）
|
||
if _budget_low:
|
||
if _can_push(code, key):
|
||
push_to_xmpp(f"⚡ {name}({code}) {price} → 进入{label}{lo}~{hi}")
|
||
outputs.append(f" 📨 区间触发(超时跳过重评)→已推送Dad")
|
||
else:
|
||
try:
|
||
cost = d.get("cost", 0) or 0
|
||
shares = d.get("shares", 0) or 0
|
||
current_action = d.get("action", "")
|
||
result = reassess_with_context(code, name, price, cost, shares, current_action)
|
||
if result:
|
||
timing_signal = result.get("timing_signal", "")
|
||
action = result.get("action", "")
|
||
# 格式化区间描述（止盈区lo=0时美化显示）
|
||
if key == "take_profit_zone" and lo == 0:
|
||
zone_desc = f"止盈监控(目标{hi:.0f})"
|
||
else:
|
||
zone_desc = f"操作区间{lo}~{hi}"
|
||
if "买入" in timing_signal or "加仓" in timing_signal or timing_signal in ("卖出","止盈"):
|
||
rr = result.get("rr_ratio", 0)
|
||
if _can_push(code, key):
|
||
msg = f"🔔 {name}({code}) 价{price}→触发{zone_desc}，已触发重评|RR={rr}"
|
||
push_to_xmpp(msg)
|
||
outputs.append(f" 📨 区间触发重评→已推送Dad: {action}")
|
||
else:
|
||
reason = f"重评结果:{timing_signal}，不构成操作建议"
|
||
outputs.append(f" 📋 本地日志(不推): {reason}")
|
||
except Exception as e:
|
||
outputs.append(f" ⚠️ 区间重评失败: {e}")
|
||
state[code][key] = True
|
||
state_updated = True
|
||
|
||
elif not in_zone and prev_in_zone == True:
|
||
if key != "stop_loss":
|
||
outputs.append(f"📌 {name}({code}) {price} → 离开{label}{lo}~{hi}")
|
||
state[code][key] = False
|
||
state_updated = True
|
||
|
||
# === 第三步：买入区偏离检测 + 自动重评 ===
|
||
reassesed_codes = []
|
||
# 先做急跌检测（仅持仓，自选股不推送暴跌告警）
|
||
holdings_codes = {d["code"] for d in active if (d.get("shares") or 0) > 0}
|
||
for d in active:
|
||
code = d["code"]
|
||
# 非持仓跳过
|
||
if code not in holdings_codes:
|
||
continue
|
||
name = d.get("name", code)
|
||
price_info = prices.get(code)
|
||
if not price_info:
|
||
continue
|
||
price, _, change_pct = price_info
|
||
if price == 0:
|
||
continue
|
||
# 单日跌幅>7%告警（不依赖zone边界，盘中急跌即触发）
|
||
try:
|
||
cp = float(change_pct) if change_pct else 0
|
||
except:
|
||
cp = 0
|
||
if cp <= -7:
|
||
prev_alert = state.get(code, {}).get("__sharp_decline_triggered", False)
|
||
if not prev_alert:
|
||
stop_loss = d.get("stop_loss", 0)
|
||
sl_note = f" 止损{stop_loss}" if stop_loss else ""
|
||
msg = f"🔻 {name}({code}) {price} 暴跌{cp:.1f}%！{sl_note}"
|
||
push_to_xmpp(msg)
|
||
outputs.append(msg)
|
||
state.setdefault(code, {})["__sharp_decline_triggered"] = True
|
||
state_updated = True
|
||
# 立即持久化，防止后续超时导致状态丢失而重复推送
|
||
save_state(state)
|
||
elif cp > -5:
|
||
# 反弹后清除告警标记，下次再跌还能报
|
||
state.setdefault(code, {}).pop("__sharp_decline_triggered", None)
|
||
|
||
for d in active:
|
||
code = d["code"]
|
||
name = d.get("name", code)
|
||
price_info = prices.get(code)
|
||
if not price_info:
|
||
continue
|
||
price, _, _ = price_info
|
||
if price == 0:
|
||
continue
|
||
|
||
# 从 decisions (DB holding_strategies) 中读取 analysis 的买入区
|
||
entry_low = d.get("entry_low", 0)
|
||
entry_high = d.get("entry_high", 0)
|
||
if not entry_low or not entry_high:
|
||
continue
|
||
|
||
in_buy_zone = entry_low <= price <= entry_high
|
||
prev_in_buy_zone = state.get(code, {}).get("__buy_zone", None)
|
||
|
||
# 状态变化时才触发
|
||
if in_buy_zone and prev_in_buy_zone == False:
|
||
# 重新进入买入区 → 重评确认区间是否仍然有效
|
||
outputs.append(f"🔄 {name}({code}) {price} → 重新进入买入区{entry_low}~{entry_high}，触发技术面重评")
|
||
do_reassess = True
|
||
elif not in_buy_zone and prev_in_buy_zone == True:
|
||
# 离开买入区 → 立即重评，更新止损/止盈/区间
|
||
outputs.append(f"🔄 {name}({code}) {price} → 离开买入区{entry_low}~{entry_high}，立即技术面重评")
|
||
do_reassess = True
|
||
else:
|
||
do_reassess = False
|
||
|
||
if do_reassess and HAS_REASSESS:
|
||
try:
|
||
cost = d.get("cost", 0) or 0
|
||
shares = d.get("shares", 0) or 0
|
||
profit_pct = (price - cost) / cost * 100 if cost else 0
|
||
is_deep_loss = profit_pct < -20
|
||
sentiment = "neutral"
|
||
if d.get("tech_snapshot"):
|
||
if "bearish" in d["tech_snapshot"]:
|
||
sentiment = "bearish"
|
||
elif "bullish" in d["tech_snapshot"]:
|
||
sentiment = "bullish"
|
||
|
||
# 调用技术面驱动重评（非机械百分比）
|
||
result = reassess_strategy(
|
||
code, name, price, cost, shares,
|
||
current_action=d.get("action", ""),
|
||
volume_signal="中性", sentiment=sentiment,
|
||
)
|
||
outputs.append(f" 📊 新策略: 损{result['stop_loss']} 盈{result['take_profit']} 区{result['entry_low']}~{result['entry_high']} RR={result['rr_ratio']}")
|
||
reassesed_codes.append(code)
|
||
except Exception as e:
|
||
outputs.append(f" ⚠️ 重评失败: {e}")
|
||
|
||
# 更新买入区状态
|
||
if "__buy_zone" not in state.get(code, {}):
|
||
if code not in state:
|
||
state[code] = {}
|
||
state[code]["__buy_zone"] = in_buy_zone
|
||
state_updated = True
|
||
|
||
# 如果有重评过的股票，更新 DB holding_strategies（此前写入 decisions.json，已废弃）
|
||
if reassesed_codes and HAS_REASSESS:
|
||
# ── 5分钟冷却：regenerate_all 开销太大，不每2分钟跑一次 ──
|
||
_regen_marker = "/tmp/price_monitor_regen_at"
|
||
_skip_regen = False
|
||
try:
|
||
if os.path.exists(_regen_marker):
|
||
with open(_regen_marker) as _f:
|
||
_last_regen = float(_f.read().strip())
|
||
if time.time() - _last_regen < 300:
|
||
_skip_regen = True
|
||
except:
|
||
pass
|
||
|
||
if _skip_regen:
|
||
outputs.append(f" ⏭ 跳过全量重评（距上次<5min），下次再跑")
|
||
else:
|
||
try:
|
||
from strategy_lifecycle import regenerate_all
|
||
r = regenerate_all(stdout=False)
|
||
outputs.append(f" ✅ 策略已全量重评: {r.get('ok',0)}/{r.get('total',0)}成功")
|
||
outputs.append(f" 📌 触发股票: {', '.join(reassesed_codes)}")
|
||
try:
|
||
with open(_regen_marker, "w") as _f:
|
||
_f.write(str(time.time()))
|
||
except:
|
||
pass
|
||
except Exception as e:
|
||
outputs.append(f" ⚠️ 全量重评失败: {e}")
|
||
|
||
# === 第四步：输出 ===
|
||
now_str = datetime.now().strftime("%H:%M:%S")
|
||
elapsed = time.time() - start
|
||
|
||
if outputs:
|
||
print(f"\n🔔 {now_str}{label}")
|
||
for o in outputs:
|
||
print(o)
|
||
print(f"\n<structured_data>{json.dumps({'type':'价格监控','time':now_str,'triggers':outputs}, ensure_ascii=False)}</structured_data>")
|
||
else:
|
||
# 无触发时 SILENT（中继不推送）
|
||
print(f"[SILENT]{label} 价格正常 | {refreshed}只已刷新 | {elapsed:.1f}s")
|
||
|
||
if state_updated:
|
||
save_state(state)
|
||
|
||
# 输出耗时
|
||
print(f"⏱{label} {elapsed:.1f}s", flush=True)
|
||
|
||
# 清理进程锁
|
||
try:
|
||
os.remove("/tmp/price_monitor.lock")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def main():
|
||
"""每cron触发跑一轮"""
|
||
run_once()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|