179 lines
6.5 KiB
Python
179 lines
6.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Shared helpers for the QMT gate scripts (qmt_sentinel.py / qmt_probe_0915.py).
|
|
|
|
Design: scripts judge AVAILABILITY (process + bridge RPC + counter behaviour),
|
|
agents/humans judge CORRECTNESS (cash baseline drift is recorded, never red).
|
|
Audit trail: C:\\sanguo_bigqmt\\gate\\qmt_gate_YYYYMMDD.log (append) +
|
|
qmt_gate_latest.json (overwrite status light).
|
|
"""
|
|
import datetime
|
|
import json
|
|
import re
|
|
import subprocess
|
|
|
|
ACCOUNT = "66639661"
|
|
GATE_DIR = r"C:\sanguo_bigqmt\gate"
|
|
RELOGIN_TASK = "sanguo-qmt-relogin"
|
|
QMT_PROCS = ("XtItClient.exe", "XtMiniQmt.exe", "miniquote.exe")
|
|
|
|
|
|
def now_str():
|
|
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def gate_log(kind, verdict, msg):
|
|
line = "%s [%s] %s %s" % (now_str(), kind, verdict, msg)
|
|
try:
|
|
import os
|
|
os.makedirs(GATE_DIR, exist_ok=True)
|
|
path = r"%s\qmt_gate_%s.log" % (
|
|
GATE_DIR, datetime.datetime.now().strftime("%Y%m%d"))
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
f.write(line + "\n")
|
|
except Exception as exc:
|
|
line += " (log-write-failed %r)" % exc
|
|
print(line, flush=True)
|
|
return line
|
|
|
|
|
|
def gate_json(kind, verdict, **detail):
|
|
import os
|
|
payload = {"ts": now_str(), "kind": kind, "verdict": verdict}
|
|
payload.update(detail)
|
|
try:
|
|
os.makedirs(GATE_DIR, exist_ok=True)
|
|
with open(r"%s\qmt_gate_latest.json" % GATE_DIR, "w",
|
|
encoding="utf-8") as f:
|
|
json.dump(payload, f, ensure_ascii=False, default=str, indent=1)
|
|
except Exception as exc:
|
|
print("GATE_JSON_WRITE_FAIL %r" % exc, flush=True)
|
|
return payload
|
|
|
|
|
|
def redis_client():
|
|
import redis
|
|
with open(r"C:\redis\redis.conf") as f:
|
|
pw = re.search(r"^requirepass (\S+)", f.read(), re.M).group(1)
|
|
return redis.Redis(host="127.0.0.1", port=6379, db=5, password=pw,
|
|
socket_connect_timeout=5, decode_responses=True)
|
|
|
|
|
|
def bridge_ping(rc, timeout=10):
|
|
"""(up, detail) — up means the RPC server answered ping with ok+pong."""
|
|
from bigqmt_signal_trader.redis_rpc import call_redis_rpc
|
|
try:
|
|
r = call_redis_rpc(rc, ACCOUNT, "ping", {}, timeout_seconds=timeout)
|
|
except Exception as exc:
|
|
return False, "RPC_DOWN %s %s" % (type(exc).__name__, str(exc)[:120])
|
|
if not isinstance(r, dict):
|
|
return False, "RPC_BAD %r" % (r,)
|
|
if not r.get("ok"):
|
|
return False, "RPC_ERR %s" % str(r.get("error") or r)[:150]
|
|
d = r.get("data") or {}
|
|
if not d.get("pong"):
|
|
return False, "RPC_NO_PONG %s" % str(d)[:150]
|
|
return True, "server_time=%s rev=%s allow_order=%s" % (
|
|
d.get("server_time"), d.get("rpc_revision"), d.get("allow_order_methods"))
|
|
|
|
|
|
def procs_alive():
|
|
"""QMT family process presence via tasklist (no psutil dependency)."""
|
|
out = subprocess.run(["tasklist", "/FO", "CSV", "/NH"],
|
|
capture_output=True, text=True, timeout=30).stdout or ""
|
|
return {name: ('"%s"' % name) in out for name in QMT_PROCS}
|
|
|
|
|
|
CALENDAR_PROBE_CODE = (
|
|
"from xtquant import xtdata; "
|
|
"r = xtdata.get_trading_dates('SH', '20260101', '20261231'); "
|
|
"e = r[-1] if isinstance(r, list) and r else None; "
|
|
"print('CAL', type(e).__name__, repr(e))"
|
|
)
|
|
|
|
|
|
def calendar_probe(timeout=25):
|
|
"""Fifth gate (added 2026-09-11): fresh child process fetches the trading
|
|
calendar through the bridge and checks the element type.
|
|
|
|
On 2026-09-11 all 12 engine schedulers starved because the bridge returned
|
|
'YYYYMMDD' STRINGS from its native xtdata path while every other probe
|
|
(process / ping / asset / probe order) stayed green. Consumers divide by
|
|
1000, so a string element is a hard contract break.
|
|
|
|
Returns (state, detail): state in ok / bad / empty / exec.
|
|
ok - non-empty numeric elements (epoch ms), contract healthy
|
|
bad - string elements -> RED (relogin restarts the patched server,
|
|
which normalizes the output again)
|
|
empty - probe shape got no data (not proof the bridge is bad) -> AMBER
|
|
exec - child process failed -> AMBER
|
|
"""
|
|
import os
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = r"C:\sanguo_bigqmt\xtquant_bridge"
|
|
env["BIGQMT_ACCOUNT_ID"] = ACCOUNT
|
|
with open(r"C:\redis\redis.conf") as f:
|
|
conf = f.read()
|
|
env["BIGQMT_REDIS_PASSWORD"] = re.search(
|
|
r"^requirepass (\S+)", conf, re.M).group(1)
|
|
try:
|
|
r = subprocess.run(
|
|
["C:\\Python310\\python.exe", "-X", "utf8", "-c",
|
|
CALENDAR_PROBE_CODE],
|
|
capture_output=True, text=True, timeout=timeout, env=env)
|
|
except Exception as exc:
|
|
return "exec", "CAL_PROBE_EXEC %s %s" % (type(exc).__name__,
|
|
str(exc)[:100])
|
|
cal = next((l for l in (r.stdout or "").splitlines()
|
|
if l.startswith("CAL ")), "")
|
|
if not cal:
|
|
return "exec", "CAL_PROBE_NO_OUT rc=%s %s" % (
|
|
r.returncode, (r.stderr or "").strip()[:100])
|
|
parts = cal.split(None, 2)
|
|
if len(parts) < 3:
|
|
return "exec", "CAL_PROBE_BAD '%s'" % cal[:80]
|
|
etype, eval_ = parts[1], parts[2][:40]
|
|
if etype == "NoneType":
|
|
return "empty", "calendar probe returned no data (%s)" % eval_
|
|
if etype in ("int", "float"):
|
|
return "ok", "calendar elem=%s %s" % (etype, eval_)
|
|
return "bad", "CALENDAR_STR_CONTRACT elem=%s %s" % (etype, eval_)
|
|
|
|
|
|
def relogin_running():
|
|
"""True if the v2 relogin python is mid-flight (avoid double trigger)."""
|
|
try:
|
|
import psutil
|
|
except ImportError:
|
|
return False
|
|
for p in psutil.process_iter(["pid", "name", "cmdline"]):
|
|
try:
|
|
if "python" not in (p.info["name"] or "").lower():
|
|
continue
|
|
cmd = " ".join(p.info["cmdline"] or [])
|
|
except Exception:
|
|
continue
|
|
if "qmt_relogin.py" in cmd:
|
|
return True
|
|
return False
|
|
|
|
|
|
def trigger_relogin():
|
|
r = subprocess.run(["schtasks", "/run", "/tn", RELOGIN_TASK],
|
|
capture_output=True, text=True, timeout=30)
|
|
ok = r.returncode == 0
|
|
return ok, ((r.stdout or "") + (r.stderr or "")).strip()[:120]
|
|
|
|
|
|
def wait_bridge(rc, timeout_sec, kind):
|
|
"""Poll bridge ping until it answers; every probe logged once at the end."""
|
|
import time
|
|
t0 = time.time()
|
|
last = ""
|
|
while time.time() - t0 < timeout_sec:
|
|
up, detail = bridge_ping(rc)
|
|
if up:
|
|
return True, int(time.time() - t0), detail
|
|
last = detail
|
|
time.sleep(5)
|
|
return False, int(time.time() - t0), last
|