# -*- coding: utf-8 -*- """ PositionJournal — 持仓期间数据记录器 每个持仓周期自动记录：订单簿快照、挂单墙、清算事件、成交量 平仓时写入 JSON 文件，供事后复盘分析 """ import json import os import time from datetime import datetime from collections import deque class PositionJournal: """单次持仓的完整数据日志""" def __init__(self, symbol, inst_id, direction, entry_price, size, leverage): self.symbol = symbol self.inst_id = inst_id self.direction = direction # LONG / SHORT self.entry_price = entry_price self.size = size self.leverage = leverage self.opened_at = time.time() self.closed_at = None # 时间序列数据 self.snapshots = [] # [{ts, price, ob_bid_wall, ob_ask_wall, spread, bid_depth_1pct, ask_depth_1pct}] self.liq_events = [] # [{ts, price, side, qty}] self.taker_flow = [] # [{ts, buy_vol, sell_vol}] self.ls_ratios = [] # [{ts, ratio}] self.alerts = [] # [{ts, level, message}] self._data_dir = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "position_journals" ) # ── 数据录入 ── def snapshot_ob(self, price, asks, bids): """记录一次订单簿快照""" ts = time.time() best_ask = float(asks[0][0]) if asks else 0 best_bid = float(bids[0][0]) if bids else 0 spread = (best_ask - best_bid) / price if price > 0 else 0 # 1%深度内挂单量 ask_1pct = sum(float(a[1]) for a in asks if float(a[0]) <= price * 1.01) bid_1pct = sum(float(b[1]) for b in bids if float(b[0]) >= price * 0.99) # 找最近的大挂单墙 (单档>5%深度) wall_threshold = (ask_1pct + bid_1pct) * 0.05 ask_walls = [(float(a[0]), float(a[1])) for a in asks if float(a[1]) >= wall_threshold and float(a[0]) <= price * 1.02] bid_walls = [(float(b[0]), float(b[1])) for b in bids if float(b[1]) >= wall_threshold and float(b[0]) >= price * 0.98] self.snapshots.append({ "ts": ts, "time": datetime.now().strftime("%H:%M:%S"), "price": round(price, 6), "spread_pct": round(spread * 100, 4), "bid_depth_1pct": round(bid_1pct, 1), "ask_depth_1pct": round(ask_1pct, 1), "ask_walls": [{"price": p, "size": s} for p, s in ask_walls], "bid_walls": [{"price": p, "size": s} for p, s in bid_walls], }) def snapshot_liq(self, price, side, qty): """记录清算事件""" self.liq_events.append({ "ts": time.time(), "time": datetime.now().strftime("%H:%M:%S"), "price": price, "side": side, "qty": qty, }) def snapshot_taker(self, buy_vol, sell_vol): """记录主动买卖量""" self.taker_flow.append({ "ts": time.time(), "time": datetime.now().strftime("%H:%M:%S"), "buy_vol": round(buy_vol, 1), "sell_vol": round(sell_vol, 1), }) def snapshot_ls_ratio(self, ratio): """记录多空比""" self.ls_ratios.append({ "ts": time.time(), "time": datetime.now().strftime("%H:%M:%S"), "ratio": round(float(ratio), 4), }) def add_alert(self, level, message): """记录告警""" self.alerts.append({ "ts": time.time(), "time": datetime.now().strftime("%H:%M:%S"), "level": level, "message": message, }) # ── 平仓分析 ── def close(self, exit_price, pnl, pnl_pct, exit_reason=""): """平仓，生成复盘摘要并写入文件""" self.closed_at = time.time() self.exit_price = exit_price self.pnl = pnl self.pnl_pct = pnl_pct self.exit_reason = exit_reason self.holding_minutes = round((self.closed_at - self.opened_at) / 60, 1) # 生成摘要 summary = self._build_summary() self._save(summary) return summary def _build_summary(self): """构建复盘摘要""" # OB快照统计 ob_stats = {} if self.snapshots: spreads = [s["spread_pct"] for s in self.snapshots] bid_depths = [s["bid_depth_1pct"] for s in self.snapshots] ask_depths = [s["ask_depth_1pct"] for s in self.snapshots] prices = [s["price"] for s in self.snapshots] ob_stats = { "snapshot_count": len(self.snapshots), "avg_spread_pct": round(sum(spreads) / len(spreads), 4), "max_spread_pct": round(max(spreads), 4), "avg_bid_depth": round(sum(bid_depths) / len(bid_depths), 1), "avg_ask_depth": round(sum(ask_depths) / len(ask_depths), 1), "price_high": round(max(prices), 6), "price_low": round(min(prices), 6), "price_range_pct": round((max(prices) - min(prices)) / prices[0] * 100, 2) if prices else 0, } # 找出最大的挂单墙 all_ask_walls = [] all_bid_walls = [] for s in self.snapshots: all_ask_walls.extend(s["ask_walls"]) all_bid_walls.extend(s["bid_walls"]) if all_ask_walls: max_ask = max(all_ask_walls, key=lambda x: x["size"]) ob_stats["max_ask_wall"] = max_ask if all_bid_walls: max_bid = max(all_bid_walls, key=lambda x: x["size"]) ob_stats["max_bid_wall"] = max_bid # 清算统计 liq_stats = {} if self.liq_events: buy_liq = [e for e in self.liq_events if e["side"] == "buy"] sell_liq = [e for e in self.liq_events if e["side"] == "sell"] liq_stats = { "total": len(self.liq_events), "buy_liq_count": len(buy_liq), "sell_liq_count": len(sell_liq), "liq_above_price": len([e for e in self.liq_events if e["price"] > self.entry_price]), "liq_below_price": len([e for e in self.liq_events if e["price"] < self.entry_price]), } # Taker流统计 taker_stats = {} if self.taker_flow: total_buy = sum(t["buy_vol"] for t in self.taker_flow) total_sell = sum(t["sell_vol"] for t in self.taker_flow) taker_stats = { "total_buy_vol": round(total_buy, 1), "total_sell_vol": round(total_sell, 1), "net_delta": round(total_buy - total_sell, 1), "buy_sell_ratio": round(total_buy / total_sell, 2) if total_sell > 0 else 0, } # L/S比率 ls_stats = {} if self.ls_ratios: ratios = [r["ratio"] for r in self.ls_ratios] ls_stats = { "avg_ratio": round(sum(ratios) / len(ratios), 4), "max_ratio": round(max(ratios), 4), "min_ratio": round(min(ratios), 4), } return { "symbol": self.symbol, "inst_id": self.inst_id, "direction": self.direction, "entry_price": self.entry_price, "exit_price": self.exit_price, "size": self.size, "leverage": self.leverage, "pnl": round(self.pnl, 8) if self.pnl else 0, "pnl_pct": round(self.pnl_pct, 6) if self.pnl_pct else 0, "exit_reason": self.exit_reason, "holding_minutes": self.holding_minutes, "opened_at": datetime.fromtimestamp(self.opened_at).strftime("%Y-%m-%d %H:%M:%S"), "closed_at": datetime.fromtimestamp(self.closed_at).strftime("%Y-%m-%d %H:%M:%S"), "ob_stats": ob_stats, "liq_stats": liq_stats, "taker_stats": taker_stats, "ls_stats": ls_stats, "alerts": self.alerts, "snapshots": self.snapshots, "liq_events": self.liq_events, "taker_flow": self.taker_flow, "ls_ratios": self.ls_ratios, } def _save(self, summary): """写入 JSON 文件""" os.makedirs(self._data_dir, exist_ok=True) fname = f"{self.symbol}_{self.direction}_{datetime.fromtimestamp(self.opened_at).strftime('%Y%m%d_%H%M%S')}.json" path = os.path.join(self._data_dir, fname) with open(path, 'w', encoding='utf-8') as f: json.dump(summary, f, ensure_ascii=False, indent=2, default=str) print(f"[Journal] Saved: {path}") return path