# -*- coding: utf-8 -*- """ 期货 EXPMA 盯盘（天勤 TqSdk 版） python monitor.py 向天勤要 5 分钟 K 线，自定义周期本地按交易日合成。 （天勤对非标准周期是按自然时钟切格的，与行情软件对不上，所以自己合成。） 启动时一次性拉够历史把 EXPMA 推到当前值，随后进入实时监听： 逼近交叉时预警，周期收盘确认交叉时立即通知。 只读行情，不含任何下单代码。 """ import os import sys from datetime import datetime from tqsdk import TqApi, TqAuth, tafunc from config import load_symbols, load_settings, LOG_DIR from expma_core import ExpmaCross from sessions import CycleBuilder, detect_night from notifier import Notifier BASE_MIN = 5 # 向天勤要的基础周期 MAX_BARS = 9800 # 单序列请求上限（官方 10000） def log(msg): now = datetime.now() line = "%s %s" % (now.strftime("%m-%d %H:%M:%S"), msg) try: print(line, flush=True) except Exception: pass try: os.makedirs(LOG_DIR, exist_ok=True) p = os.path.join(LOG_DIR, now.strftime("%Y-%m-%d") + ".log") with open(p, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: pass class Watch(object): """单个合约的盯盘单元""" def __init__(self, cfg, notifier): self.symbol = cfg["symbol"] self.name = cfg["name"] self.period = int(cfg["period"]) self.warn_pct = float(cfg.get("warn_pct", 0.3)) self.cross = ExpmaCross(cfg["fast"], cfg["slow"]) self.notifier = notifier self.klines = None self.builder = None self.has_night = None self.last_price = None self.last_cycle_dt = None self._warned = set() self._live = False def need_bars(self, multiple=3): """预热需要的 5 分钟 K 线根数""" cycles = int(self.cross.slow_n * multiple * 1.2) + 30 return min(cycles * (self.period // BASE_MIN), MAX_BARS) # ---------- 预热 ---------- def warmup(self): kl = self.klines closed = kl.iloc[:-1] dts = [tafunc.time_to_datetime(x) for x in closed["datetime"].values] self.has_night = detect_night(dts) self.builder = CycleBuilder(self.period, BASE_MIN, self.has_night) n5 = ncyc = 0 for dt, o, h, l, c in zip(dts, closed["open"].values, closed["high"].values, closed["low"].values, closed["close"].values): if c != c: continue n5 += 1 done = self.builder.push(dt, float(o), float(h), float(l), float(c)) if done: self.cross.update(done["close"]) self.last_cycle_dt = done["datetime"] ncyc += 1 self._live = True need = self.cross.slow_n * 3 ok = self.cross.ready() log(" %-16s %-10s EXPMA(%d,%d) %d分钟 %s" % (self.symbol, self.name, self.cross.fast_n, self.cross.slow_n, self.period, "有夜盘" if self.has_night else "无夜盘")) log(" 5分钟%d根 -> 周期K线%d根 快%.2f 慢%.2f %s%s" % (n5, ncyc, self.cross.fast or 0, self.cross.slow or 0, "多头" if (self.cross.fast or 0) > (self.cross.slow or 0) else "空头", "" if ok else " ⚠未收敛(需%d根)" % need)) if self.last_cycle_dt: log(" 最后一根周期K线起于 %s" % self.last_cycle_dt.strftime("%m-%d %H:%M")) return ncyc # ---------- 5 分钟 K 线收盘 ---------- def on_5min_closed(self, bar): c = bar["close"] if c != c: return dt = tafunc.time_to_datetime(bar["datetime"]) done = self.builder.push(dt, float(bar["open"]), float(bar["high"]), float(bar["low"]), float(c)) if done: self.on_cycle_closed(done) # ---------- 周期收盘 ---------- def on_cycle_closed(self, bar): sig = self.cross.update(bar["close"]) self.last_cycle_dt = bar["datetime"] self._warned.clear() if not sig or not self._live: return kind = "金叉 ↑" if sig == "gold" else "死叉 ↓" title = "【期货】%s %s %s" % (self.name, self.symbol, kind) content = ( "合约：%s (%s)\n" "信号：EXPMA %s（周期已收盘确认）\n" "周期：%d 分钟 参数：快 %d / 慢 %d\n" "快线：%.2f\n慢线：%.2f\n收盘：%.2f\n" "K线：%s ~ %s" % (self.name, self.symbol, kind, self.period, self.cross.fast_n, self.cross.slow_n, self.cross.fast, self.cross.slow, bar["close"], bar["datetime"].strftime("%Y-%m-%d %H:%M"), bar["end"].strftime("%H:%M")) ) log("★ " + title) self.notifier.send(title, content) # ---------- 盘中预警 ---------- def check_warn(self, price): if not self._live or not self.cross.ready() or not price: return cp = self.cross.cross_price() if not cp or cp <= 0: return crossed = False direction = "gold" if cp > price else "dead" tent = self.cross.tentative(price) if tent and self.cross.prev_diff is not None: td = tent[0] - tent[1] if self.cross.prev_diff <= 0 < td or self.cross.prev_diff >= 0 > td: crossed = True direction = "gold" if td > 0 else "dead" gap_pct = abs(cp - price) / price * 100.0 if not crossed and gap_pct > self.warn_pct: return level = "crossed" if crossed else "near" key = (direction, level) if key in self._warned: return self._warned.add(key) kind = "金叉" if direction == "gold" else "死叉" tag = "盘中已穿越，待收盘确认" if crossed else "逼近" left = self.builder.remaining * BASE_MIN if self.builder else 0 title = "【期货预警】%s %s %s%s" % (self.name, self.symbol, tag, kind) content = ( "合约：%s (%s)\n" "状态：%s%s（未确认，本根周期没走完）\n" "现价：%.2f\n%s临界价：%.2f（相差 %+.2f%%）\n" "已确认快线：%.2f\n已确认慢线：%.2f\n" "周期：%d 分钟，本根还剩约 %d 分钟\n" "时间：%s" % (self.name, self.symbol, tag, kind, price, kind, cp, (cp - price) / price * 100.0, self.cross.fast, self.cross.slow, self.period, left, datetime.now().strftime("%m-%d %H:%M:%S")) ) log("⚠ " + title) self.notifier.send(title, content) def main(): print("=" * 62) print("期货 EXPMA 盯盘（天勤版） %s" % datetime.now().strftime("%Y-%m-%d %H:%M:%S")) print("=" * 62) try: settings = load_settings() symbols = load_symbols() except Exception as e: print("配置读取失败: %s" % e) return 1 tq = settings.get("tqsdk") or {} user, pwd = tq.get("user", ""), tq.get("password", "") if not user or not pwd or "填" in str(user): print("\n请先在 settings.json 里填快期账户：") print(' "tqsdk": {"user": "手机号或邮箱", "password": "密码"}') print("\n免费注册：https://account.shinnytech.com/") return 1 notifier = Notifier(settings, log) watches = [Watch(c, notifier) for c in symbols] log("连接天勤行情服务器 ...") try: api = TqApi(auth=TqAuth(user, pwd)) except Exception as e: print("\n登录失败: %s" % e) return 1 log("已连接") try: log("订阅 %d 个合约的 5 分钟 K 线 ..." % len(watches)) for w in watches: w.klines = api.get_kline_serial(w.symbol, BASE_MIN * 60, data_length=w.need_bars()) api.wait_update() log("预热指标 ...") for w in watches: w.warmup() log("盯盘中（含夜盘，行情驱动）") while True: api.wait_update() for w in watches: kl = w.klines cur = kl.iloc[-1] if api.is_changing(cur, "datetime"): w.on_5min_closed(kl.iloc[-2]) if api.is_changing(cur, "close"): c = cur["close"] if c == c: w.last_price = float(c) w.check_warn(float(c)) except KeyboardInterrupt: log("收到中断") except Exception as e: log("异常退出: %s: %s" % (type(e).__name__, e)) return 1 finally: try: api.close() except Exception: pass log("已停止") return 0 if __name__ == "__main__": sys.exit(main())