diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index c64b364..bedda59 100644 --- a/py-client/strategy/zt/boot.py +++ b/py-client/strategy/zt/boot.py @@ -1,112 +1,114 @@ -"""日内做 T 策略启动器。""" +"""做 T 策略启动器。 + +该模块负责组合 SDK、配置、状态存储和做 T 策略组件，供 main.py 调用。 +""" from __future__ import annotations -import logging +import logging as log import time -from concurrent.futures import Future, ThreadPoolExecutor from datetime import datetime, time as clock_time import config from libs.calc import trading_time -from libs.grid_take_profit import GridTrailingTracker from libs.market import market_allow_open from libs.signal import init_signals +from libs.collector import collector_push +from libs.grid_take_profit import GridTrailingTracker from sdk import Client -from strategy.trend.order import OrderBook -from strategy.trend.watch import DipWatch - +from .order import OrderBook +from .watch import DipWatch +from .runtime import Runtime +from .state import TState, SOLD from .open import open_signal from .positions import manage_positions -from .runtime import Runtime -from .state import TState def StartZT() -> None: - client = Client(config.global_config.qmt_base_url, config.global_config.qmt_token, config.HTTP_TIMEOUT) - orders = OrderBook() - orders.refresh(client) - _, positions = client.positions() - state = TState.for_strategy(config.global_config.qmt_data_dir, config.account_config.strategy, config.account_config.account_id) - state.reconcile(positions, orders.data, datetime.now().date().isoformat()) - run = Runtime(client, config.global_config, config.account_config, state, orders, DipWatch(), GridTrailingTracker(config.account_config.grid_step_pct)) - while True: - started = time.monotonic() - try: - RunOnce(run) - except Exception: - logging.exception("ZT 策略本轮失败") - time.sleep(max(0.0, 30.0 - (time.monotonic() - started))) + """初始化做 T 策略，并以 30 秒间隔持续执行。""" + with Client(config.global_config.qmt_base_url, config.global_config.qmt_token, + config.HTTP_TIMEOUT) as client: + state = TState.for_strategy(config.global_config.qmt_data_dir, "zt", config.account_config.account_id) + run = Runtime(client, config.global_config, config.account_config, state, + OrderBook(), DipWatch(), DipWatch(), + GridTrailingTracker(config.account_config.grid_step_pct)) + log.info("[ZT 启动] 账户=%s，底仓信号=dcm，状态文件=%s", run.account_cfg.account_id, state.path) + while True: + now = datetime.now() + if now.time() >= clock_time(15): + # 收盘前最后一次只读对账，不发新单；未完成买回继续持久保存。 + try: + portfolio = client.portfolio() + state.reconcile(list(portfolio.positions.values()), portfolio.orders, now.date().isoformat()) + except Exception: + log.exception("[ZT] 收盘对账失败，保留本地待确认记录") + for item in state.items.values(): + if item.phase == SOLD or state.busy(item.code): + log.warning("[ZT] 收盘仍有待完成轮次：%s", item.code) + return + # 单轮失败不能杀死唯一的交易定时线程。 + try: + RunOnce(run) + except Exception: + log.exception("[ZT] 本 tick 执行失败，下一个 tick 继续") + # 计算距离下一个目标时间点（0秒或30秒）的等待时间。 + time.sleep(30 - datetime.now().second % 30) def RunOnce(run: Runtime) -> None: - if not trading_time(datetime.now()): + """账户快照 → 成交对账 → 做 T 管理 → dcm 建仓，共用一份资金预算。""" + now = datetime.now() + if not trading_time(now) or now.time() >= clock_time(15): return + today = now.date().isoformat() + started_at = time.monotonic() + + # 1. 一次获取资产、持仓和订单，并清理过期订单。 + portfolio = run.client.portfolio() + positions = list(portfolio.positions.values()) + run.orders.refresh(run.client, portfolio.orders) + # 对账使用完整原始订单列表，不能丢弃撤单和废单的部分成交。 + run.state.reconcile(positions, portfolio.orders, today) + + # 2. 获取本策略的信号开仓数据；信号失败不阻断已有做 T 买回。 try: - run.orders.refresh(run.client) - assets = run.client.assets() - position_codes, positions = run.client.positions() + signals = init_signals(run.global_cfg, ["dcm"]) except Exception: - logging.exception("[ZT] 刷新账户或订单失败") + log.exception("[ZT] 获取 dcm 信号失败，本轮只管理已有底仓") + signals = [] + position_codes = {position.stock_code for position in positions if position.volume > 0} + candidates = [signal for signal in signals + if signal.signal_key == "dcm" and signal.code not in position_codes] + + # 3. 获取持仓和待开仓证券的实时行情 tick，零持仓的待买回证券也包含在内。 + codes = list(dict.fromkeys(list(position_codes) + list(run.state.items) + + [signal.code for signal in candidates])) + ticks = run.client.full_tick(codes) if codes else {} + now = datetime.now() # 网络请求可能跨过尾盘边界，提交前重新判断。 + if not trading_time(now) or now.time() >= clock_time(15): return - today = datetime.now().date().isoformat() - try: - signals = init_signals(run.global_cfg, run.account_cfg.signal_allow) - except Exception: - logging.exception("[ZT] 获取 dcm 信号失败") - return - candidate_codes = [item.code for item in signals if item.code not in position_codes] - codes = list(dict.fromkeys(position_codes + candidate_codes)) - try: - ticks = run.client.full_tick(codes) - except Exception: - logging.exception("[ZT] 获取行情失败") - return - market_ok = market_allow_open(run.global_cfg.api_host) - can_open = market_ok and assets.available >= assets.total * run.account_cfg.min_cash_ratio - force_buy_back = datetime.now().time() >= clock_time(14, 50) - # 状态对账与开仓判断并行。持仓线程在自己的线程中等待对账完成， - # 以保证它读取到最新的底仓和做 T 轮次状态，避免并发写 State。 - with ThreadPoolExecutor(max_workers=3, thread_name_prefix="zt") as executor: - state_future = executor.submit(run.state.reconcile, positions, run.orders.data, today) - open_future = executor.submit(_run_open_signal, state_future, run, ticks, signals, can_open) - positions_future = executor.submit( - _run_manage_positions, - state_future, - run, - ticks, - positions, - assets.available, - today, - force_buy_back, - ) - _wait_worker("状态对账", state_future) - _wait_worker("开仓", open_future) - _wait_worker("持仓管理", positions_future) + # 4. 先完成买回，避免开底仓抢占资金；交易逻辑串行，状态无需多线程写入。 + available = max(0.0, portfolio.assets.available) + # 未确认买单可能尚未反映在资金快照中，保守预留，宁可少买也不重复使用。 + for pending in run.state.pending.values(): + if pending.kind != "sell": + tick = ticks.get(pending.code) + if tick is None or tick.last_price <= 0: + available = 0.0 + break + available = max(0.0, available - pending.qty * tick.last_price * 1.01) + force = now.time() >= clock_time(14, 50) + available = manage_positions(run, ticks, positions, available, today, force) + # 5. 验证可用资金；低于资金安全线时禁止开新仓，尾盘只完成做 T 买回。 + reserve = max(0.0, portfolio.assets.total * run.account_cfg.min_cash_ratio) + # 未完成的卖出/买回可能继续占用资金，不再额外开底仓。 + outstanding = bool(run.state.pending) or any(item.phase == SOLD for item in run.state.items.values()) + if not force and not outstanding and market_allow_open() and available > reserve: + open_signal(run, ticks, candidates, available - reserve) -def _run_open_signal(state_future: Future, run: Runtime, ticks, signals, can_open: bool) -> None: - state_future.result() - if can_open: - open_signal(run, ticks, signals) - - -def _run_manage_positions( - state_future: Future, - run: Runtime, - ticks, - positions, - available: float, - today: str, - force_buy_back: bool, -) -> None: - state_future.result() - manage_positions(run, ticks, positions, available, today, force_buy_back) - - -def _wait_worker(name: str, future: Future) -> None: - try: - future.result() - except Exception: - logging.exception("[ZT] %s线程失败", name) + # 6. 数据采集不与交易逻辑争用状态；采集函数自身隔离传输异常。 + collector_push(run.account_cfg.account_id, portfolio.assets, positions) + log.info("[ZT] 本轮完成，底仓=%d，待确认=%d，耗时=%d毫秒", + len(run.state.items), len(run.state.pending), int((time.monotonic() - started_at) * 1000)) diff --git a/py-client/strategy/zt/open.py b/py-client/strategy/zt/open.py index 6cdfed8..7b832a1 100644 --- a/py-client/strategy/zt/open.py +++ b/py-client/strategy/zt/open.py @@ -2,30 +2,100 @@ from __future__ import annotations -import logging +from datetime import datetime +import logging as log +import math from libs.calc import calc_buy_volume from sdk import OP_BUY -from strategy.trend.order import PlaceOrderRequest +from .runtime import Runtime +from .order import PlaceOrderRequest +from .state import PendingOrder -def open_signal(run, ticks, signals) -> None: - """仅处理 dcm 信号，使用趋势策略同款反弹确认建立底仓。""" - for signal in signals: - if signal.signal_key != "dcm" or run.orders.busy(signal.code, "BUY"): +def open_signal(run: Runtime, ticks, signals, available: float) -> float: + """逐个验证开仓信号并提交买入委托，返回本轮剩余资金。""" + for item in signals: + try: + now = datetime.now() + if (now.hour, now.minute) >= (14, 50): + break + if item.signal_key != "dcm" or item.code in run.account_cfg.excluded_codes: + continue + state = run.state.items.get(item.code) + if state is not None and state.base_qty > 0: + continue + # 1. 验证信号配置允许开仓的时间区间。 + signal_config = run.global_cfg.signals.get("dcm") + if signal_config is None or not check_timezone(signal_config.timezone): + continue + # 2. 检查该证券是否已有买入委托锁，防止重复下单。 + if run.state.busy(item.code) or run.orders.busy(item.code, "BUY") or run.orders.busy(item.code, "SELL"): + continue + # 3. 验证行情和最新价格是否有效。 + tick = ticks.get(item.code) + price = tick.last_price if tick else 0.0 + if not math.isfinite(price) or price <= 0 or price > run.account_cfg.zt_max_price: + continue + # 4. 根据单笔买入金额计算整手开仓数量，预留少量价差和费用。 + budget = min(run.account_cfg.buy_value, available) + volume = calc_buy_volume(price, budget) + amount = price * volume * 1.01 + if volume <= 0 or price * volume > budget or amount > available: + continue + # 5. 等待价格从观察低点反弹，防止直接接下跌中的“飞刀”。 + if not run.open_watch.triggered("ZT 建仓", item.code, price): + continue + order_id = run.orders.new_order_id("base") + request = PlaceOrderRequest(OP_BUY, item.code, volume, order_id, "zt", kind="base") + run.state.new_order(PendingOrder(order_id, item.code, "base", volume, datetime.now().date().isoformat())) + # 即使响应丢失，也保留资金预算和 pending，不能继续使用这笔钱。 + available -= amount + if run.orders.place(run.client, request): + run.open_watch.forget(item.code) + log.info("[ZT 建仓] %s 买入 %d 股，等待实际成交", item.code, volume) + except Exception: + log.exception("[ZT 建仓] %s 处理异常，继续后续信号", item.code) + return available + + +def check_timezone(timezone: str, now: datetime | None = None) -> bool: + """验证当前时间是否处于配置区间。 + + ``*`` 表示全天允许；多个区间用逗号分隔，例如 + ``9:30-10:30,13:30-14:30``。同时支持跨午夜区间。 + """ + timezone = str(timezone or "").strip() + if timezone == "*": + return True + + current = now or datetime.now() + current_minutes = current.hour * 60 + current.minute + + for section in timezone.split(","): + bounds = section.strip().split("-") + if len(bounds) != 2: continue - tick = ticks.get(signal.code) - price = tick.last_price if tick else 0.0 - if price <= 0 or price > run.account_cfg.zt_max_price: + start = _parse_minutes(bounds[0]) + end = _parse_minutes(bounds[1]) + if start is None or end is None: continue - volume = calc_buy_volume(price, run.account_cfg.buy_value) - if volume <= 0 or not run.buy_watch.triggered("ZT 建仓", signal.code, price): - continue - request = PlaceOrderRequest(run.client, OP_BUY, signal.code, volume, run.orders.new_order_id("base"), run.account_cfg.strategy) - if run.orders.place(request): - run.buy_watch.forget(signal.code) - logging.info("[ZT 建仓] %s 买入 %d 股", signal.code, volume) + + if start <= end and start <= current_minutes <= end: + return True + if start > end and (current_minutes >= start or current_minutes <= end): + return True + + return False -# 与 trend 策略的开仓函数命名保持一致。 -open_base = open_signal +def _parse_minutes(value: str) -> int | None: + """把 ``时:分`` 转换为当天分钟数，无效值返回 None。""" + try: + hour_text, minute_text = value.strip().split(":") + hour, minute = int(hour_text), int(minute_text) + except (TypeError, ValueError): + return None + if not 0 <= hour <= 23 or not 0 <= minute <= 59: + return None + return hour * 60 + minute diff --git a/py-client/strategy/zt/order.py b/py-client/strategy/zt/order.py new file mode 100644 index 0000000..1c39e49 --- /dev/null +++ b/py-client/strategy/zt/order.py @@ -0,0 +1,125 @@ +"""做 T 策略委托簿，对应 Go 客户端的 ``logic/order.go``。""" + +from __future__ import annotations + +import secrets +from dataclasses import dataclass +from datetime import datetime, timedelta +from threading import Lock +from cachelib import SimpleCache +import logging + +import httpx +from sdk import Client,ORDER_SIDE_BY_OFFSET,APIError,OrderItem + +# 表示委托仍在处理、可能继续成交的 QMT 状态。 +BUSY_STATUSES = {"48", "49", "50", "51", "52", "55"} +COMPLETED_STATUSES = {"56"} +TRACKED_STATUSES = BUSY_STATUSES | COMPLETED_STATUSES +CANCELABLE_STATUSES = {"49", "50", "51", "52"} + + +@dataclass(slots=True) +class PlaceOrderRequest: + """``OrderBook.place`` 提交委托所需的全部参数。""" + op: int + code: str + volume: int + order_id: str + strategy_name: str + kind: str = "" + + +class OrderBook: + """线程安全的活动委托缓存。""" + + def __init__(self, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 10) -> None: + self.lock_timeout_sec = max(1, lock_timeout_sec) + self.cancel_timeout_sec = timedelta(seconds=cancel_timeout_sec) + self.data: list[OrderItem] = [] + self.busy_keys: set[str] = set() + self.busy_cache = SimpleCache(threshold=10_000, default_timeout=self.lock_timeout_sec) + self.mutex = Lock() + + @staticmethod + def new_order_id(side:str) -> str: + """生成 ``zt-xxxxxxxx`` 格式的本地订单号。""" + return f"zt-{side}-{secrets.token_hex(10)}" + + def busy(self, code: str, side: str) -> bool: + """判断证券是否存在仍在处理中的同方向委托。""" + with self.mutex: + key = self._busy_key(side, code) + return key in self.busy_keys or self.busy_cache.has(key) + + @staticmethod + def _busy_key(side: str, code: str) -> str: + return f"{side}-{code}" + + def refresh(self, client: Client, orders: list[OrderItem]) -> None: + """用账户快照刷新委托，并撤销超时的活动委托。""" + current = datetime.now() + data: list[OrderItem] = [] + busy_keys: set[str] = set() + canceled = 0 + + for item in orders: + # 不处理状态不对的 + if item.status not in TRACKED_STATUSES: + continue + if item.status in BUSY_STATUSES: + busy_keys.add(self._busy_key(item.side, item.code)) + # 清理过期的 + if ( + item.created_at is not None + and item.local_order_id.startswith("zt-") + and item.status in CANCELABLE_STATUSES + and current - item.created_at > self.cancel_timeout_sec + ): + try: + client.cancel_by_id(item.id) + canceled += 1 + logging.info("[Order] 超时撤单，代码=%s，方向=%s，柜台订单=%s", item.code, item.side, item.id) + except Exception: + logging.exception("[Order] 撤单失败，保留在途状态，订单=%s", item.id) + + # 缓存本次有效订单 + data.append(item) + + with self.mutex: + self.data = data + self.busy_keys = busy_keys + logging.info("[Order] 刷新完成，跟踪=%d，处理中=%d，撤销=%d", len(data), len(busy_keys), canceled) + + def place(self, client: Client, request: PlaceOrderRequest) -> bool: + """按最新价提交委托，并立即写入本地方向锁。""" + side = ORDER_SIDE_BY_OFFSET.get(str(request.op), "") + if not side: + logging.warning("[Order] 下单失败，代码=%s，原因=未知买卖方向(%s)", request.code, request.op) + return False + + key = self._busy_key(side, request.code) + with self.mutex: + if key in self.busy_keys or self.busy_cache.has(key): + logging.info("[Order] 跳过重复下单，代码=%s，方向=%s", request.code, side) + return False + self.busy_cache.set(key, True, timeout=self.lock_timeout_sec) + + try: + result = client.passorder( + op_type=request.op, + stock_code=request.code, + volume=request.volume, + strategy_name=request.strategy_name, + order_id=request.order_id, + ) + except APIError as exc: + logging.exception("[Order] 下单失败，代码=%s，本地订单=%s，HTTP状态=%d，错误=%s", request.code, request.order_id, exc.status_code, exc.message or str(exc)) + return False + except (httpx.RequestError, ValueError): + # 响应异常不能证明柜台未受理，保留缓存防重，不自动重试。 + logging.exception("[Order] 下单请求或响应异常，代码=%s，本地订单=%s", request.code, request.order_id) + return False + + logging.info("[Order] 下单已受理，代码=%s，方向=%s，数量=%d，本地订单=%s，返回=%s", request.code, side, request.volume, request.order_id, result) + return True diff --git a/py-client/strategy/zt/positions.py b/py-client/strategy/zt/positions.py index b6adc4a..8e5ac59 100644 --- a/py-client/strategy/zt/positions.py +++ b/py-client/strategy/zt/positions.py @@ -1,71 +1,94 @@ -"""日内先卖后买的做 T 规则。""" +"""日内先卖后买的做 T 规则，不包含趋势补仓或整仓止盈。""" from __future__ import annotations -import logging +from datetime import datetime +import logging as log +import math from libs.grid_take_profit import GridState from sdk import OP_BUY, OP_SELL, PositionItem -from strategy.trend.order import PlaceOrderRequest - -from .state import BUYING, READY, SELLING, SOLD +from .order import PlaceOrderRequest +from .runtime import Runtime +from .state import PendingOrder, READY, SOLD -def manage_positions(run, ticks, positions: list[PositionItem], available: float, today: str, force_buy_back: bool = False) -> None: - for position in positions: - code = position.stock_code - tick = ticks.get(code) - if not code or code in run.account_cfg.excluded_codes or tick is None: - continue - price = tick.last_price - if price <= 0 or price > run.account_cfg.zt_max_price: - continue +def manage_positions(run: Runtime, ticks, positions: list[PositionItem], available: float, + today: str, force_buy_back: bool = False) -> float: + """遍历本地底仓记录；全部卖出后即使持仓快照为空，也必须处理买回。""" + by_code = {position.stock_code: position for position in positions} + for code, state in list(run.state.items.items()): try: - state = run.state.get(code) - except KeyError: - continue - if state.phase == READY and not force_buy_back: - _try_sell(run, state, position, price, today) - elif state.phase == SOLD: - _try_buy_back(run, state, price, available, today, force_buy_back) + now = datetime.now() + if now.hour >= 15: + break + force_buy_back = force_buy_back or (now.hour, now.minute) >= (14, 50) + if code in run.account_cfg.excluded_codes or run.state.busy(code): + continue + if run.orders.busy(code, "BUY") or run.orders.busy(code, "SELL"): + continue + tick = ticks.get(code) + price = tick.last_price if tick else 0.0 + if not math.isfinite(price) or price <= 0: + continue + position = by_code.get(code) + actual_qty = position.volume if position else 0 + expected_qty = state.base_qty - state.sell_qty + state.buy_qty + # 快照延迟或手动增减仓不能当作新的做 T 信号，先核对数量差异。 + if actual_qty != expected_qty: + log.warning("[ZT 持仓] %s 数量不符，记录=%d，实际=%d，暂停交易", code, expected_qty, actual_qty) + continue + if state.phase == SOLD: + available = _try_buy_back(run, state, price, available, today, force_buy_back) + elif state.phase == READY and position and not force_buy_back: + if price <= run.account_cfg.zt_max_price: + _try_sell(run, state, position, price, today) + except Exception: + log.exception("[ZT 持仓] %s 处理异常，继续后续证券", code) + return available -def _try_sell(run, state, position: PositionItem, price: float, today: str) -> None: - if state.base_cost <= 0 or run.orders.busy(state.code, "SELL"): +def _try_sell(run: Runtime, state, position: PositionItem, price: float, today: str) -> None: + """基于独立保存的底仓成本，用跨轮最高盈利网格判断做 T 卖出。""" + if state.base_cost <= 0: return pnl_rate = (price - state.base_cost) / state.base_cost * 100 - observation = run.sell_tracker.observe(f"{run.account_cfg.account_id}:{state.code}", pnl_rate) + key = f"{run.account_cfg.account_id}:{state.code}:{today}" + observation = run.sell_tracker.observe(key, pnl_rate) if observation.state != GridState.RETREAT: return - volume = min(position.can_use_volume, int(state.base_qty * run.account_cfg.zt_sell_ratio) // 100 * 100) + volume = min(position.can_use_volume, int(state.base_qty * run.account_cfg.zt_sell_ratio)) + volume = volume // 100 * 100 if volume <= 0: return order_id = run.orders.new_order_id("t-sell") - request = PlaceOrderRequest(run.client, OP_SELL, state.code, volume, order_id, run.account_cfg.strategy) - if not run.orders.place(request): - return - state.trade_date, state.phase = today, SELLING - state.sell_order_id, state.sell_qty, state.sell_price = order_id, volume, price - run.state.set(state) - run.state.save() - logging.info("[ZT 卖出] %s %d 股，网格回撤触发", state.code, volume) + request = PlaceOrderRequest(OP_SELL, state.code, volume, order_id, "zt", kind="sell") + run.state.new_order(PendingOrder(order_id, state.code, "sell", volume, today)) + if run.orders.place(run.client, request): + log.info("[ZT 卖出] %s %d 股，等待成交后确定买回数量和价格", state.code, volume) -def _try_buy_back(run, state, price: float, available: float, today: str, force: bool) -> None: +def _try_buy_back(run: Runtime, state, price: float, available: float, today: str, force: bool) -> float: + """按实际卖出均价下跌后反弹买回；尾盘不再受下跌幅度、反弹及价格上限限制。""" target = state.sell_price * (1 - run.account_cfg.zt_buy_fall_pct / 100) - if (not force and price > target) or run.orders.busy(state.code, "BUY"): - return - if state.sell_qty <= 0 or price * state.sell_qty > available: - return + if not force and (price > target or price > run.account_cfg.zt_max_price): + return available + volume = state.sell_qty - state.buy_qty + amount = price * volume * 1.01 + if volume <= 0 or amount > available: + log.warning("[ZT 买回] %s 买回资金不足或数量无效，保留未完成轮次", state.code) + return available if not force and not run.buy_watch.triggered("ZT 买回", state.code, price): - return + return available order_id = run.orders.new_order_id("t-buy") - request = PlaceOrderRequest(run.client, OP_BUY, state.code, state.sell_qty, order_id, run.account_cfg.strategy) - if not run.orders.place(request): - return - state.trade_date, state.phase, state.buy_order_id = today, BUYING, order_id - run.state.set(state) - run.state.save() - run.buy_watch.forget(state.code) - reason = "尾盘强制买回" if force else f"回撤 {run.account_cfg.zt_buy_fall_pct:.2f}% 后反弹确认" - logging.info("[ZT 买回] %s %d 股，%s", state.code, state.sell_qty, reason) + request = PlaceOrderRequest(OP_BUY, state.code, volume, order_id, "zt", kind="buy") + run.state.new_order(PendingOrder(order_id, state.code, "buy", volume, today)) + # pending 已落盘，任何请求结果都预留资金；下一轮再从柜台快照确认。 + available -= amount + try: + if run.orders.place(run.client, request): + run.buy_watch.forget(state.code) + log.info("[ZT 买回] %s %d 股，%s", state.code, volume, "尾盘强制买回" if force else "下跌后反弹") + except Exception: + log.exception("[ZT 买回] %s 请求结果未知，保留 pending 和预算", state.code) + return available diff --git a/py-client/strategy/zt/runtime.py b/py-client/strategy/zt/runtime.py index cc86012..33831f0 100644 --- a/py-client/strategy/zt/runtime.py +++ b/py-client/strategy/zt/runtime.py @@ -1,22 +1,31 @@ -"""做 T 策略的运行期依赖。""" +"""做 T 策略单次运行所需的上下文对象。""" from dataclasses import dataclass from config import AccountConfig, GlobalConfig from libs.grid_take_profit import GridTrailingTracker from sdk import Client -from strategy.trend.order import OrderBook -from strategy.trend.watch import DipWatch - +from .order import OrderBook +from .watch import DipWatch from .state import TState @dataclass(slots=True) class Runtime: + """集中保存策略运行期间共享的依赖和状态。 + + 将这些对象集中到一个 dataclass 后，开仓、持仓管理和单轮调度函数 + 只需接收一个 Runtime，无需重复传递大量参数。 + """ + + # 外部服务与账户配置。 client: Client global_cfg: GlobalConfig account_cfg: AccountConfig + + # 策略运行过程中共享的状态组件。 state: TState orders: OrderBook + open_watch: DipWatch buy_watch: DipWatch sell_tracker: GridTrailingTracker diff --git a/py-client/strategy/zt/state.py b/py-client/strategy/zt/state.py index 84de7b1..538b4aa 100644 --- a/py-client/strategy/zt/state.py +++ b/py-client/strategy/zt/state.py @@ -1,20 +1,18 @@ -"""做 T 策略的底仓与日内轮次状态。""" +"""做 T 策略的底仓、待确认委托和实际成交记录。""" from __future__ import annotations import json -from dataclasses import asdict, dataclass +import logging as log +import math +from dataclasses import asdict, dataclass, field from pathlib import Path -from threading import Lock -from typing import Iterable +from time import time from sdk import OrderItem, PositionItem -READY = "READY" -SELLING = "SELLING" -SOLD = "SOLD" -BUYING = "BUYING" -DONE = "DONE" +READY, SOLD, DONE = "READY", "SOLD", "DONE" +TERMINAL_STATUSES = {"53", "54", "56", "57"} @dataclass(slots=True) @@ -28,76 +26,139 @@ class TStateItem: sell_qty: int = 0 sell_price: float = 0.0 buy_order_id: str = "" + base_order_id: str = "" + buy_qty: int = 0 + buy_cost: float = 0.0 + + +@dataclass(slots=True) +class PendingOrder: + order_id: str + code: str + kind: str # base：底仓；sell：做 T 卖出；buy：做 T 买回 + qty: int + trade_date: str + submit_at: float = field(default_factory=time) class TState: - """持久化 dcm 底仓和每只证券每日一次的做 T 进度。""" + """交易逻辑串行更新；JSON 保存底仓、待确认委托及成交历史。""" def __init__(self, path: str | Path) -> None: self.path = Path(path) - self.lock = Lock() - self.items = self._load() + self.items: dict[str, TStateItem] = {} + self.pending: dict[str, PendingOrder] = {} + self.records: list[dict] = [] + self._load() @classmethod - def for_strategy(cls, data_dir: str | Path, strategy: str, account_id: str) -> "TState": + def for_strategy(cls, data_dir: str | Path, strategy: str, account_id: str) -> TState: return cls(Path(data_dir) / f"{strategy}_{account_id}_state.json") - def get(self, code: str) -> TStateItem: - with self.lock: - return self.items[code] + def busy(self, code: str) -> bool: + return any(order.code == code for order in self.pending.values()) - def set(self, item: TStateItem) -> None: - with self.lock: - self.items[item.code] = item + def new_order(self, order: PendingOrder) -> None: + """下单前落盘；请求超时不能当作失败删除，等待后续委托确认。""" + if self.busy(order.code): + raise ValueError(f"{order.code} 已有待确认委托") + self.pending[order.order_id] = order + try: + self.save() + except Exception: + del self.pending[order.order_id] + raise - def reconcile(self, positions: Iterable[PositionItem], orders: list[OrderItem], today: str) -> None: - position_list = [item for item in positions if item.stock_code and item.volume > 0] - position_codes = {item.stock_code for item in position_list} - by_local_id: dict[str, list[OrderItem]] = {} + def reconcile(self, positions: list[PositionItem], orders: list[OrderItem], today: str) -> None: + """先按实际成交记账，再接管未知持仓；不覆盖已记录的底仓成本。""" + # 同一本地委托可能拆单；按券商订单号去重，数量齐全且全部结束才记账。 + by_id: dict[str, dict[str, OrderItem]] = {} for order in orders: if order.local_order_id: - by_local_id.setdefault(order.local_order_id, []).append(order) - - for position in position_list: - if position.stock_code not in self.items and position.open_price > 0: - self.set(TStateItem(position.stock_code, position.volume, position.open_price)) - - for code in list(self.items): - item = self.get(code) - if code not in position_codes: - with self.lock: - self.items.pop(code, None) + by_id.setdefault(order.local_order_id, {})[order.id] = order + for order_id, pending in list(self.pending.items()): + side = "SELL" if pending.kind == "sell" else "BUY" + rows = [row for row in by_id.get(order_id, {}).values() + if row.code == pending.code and row.side == side] + if not rows: + log.warning("[ZT 状态] 委托暂未查到，保留待确认：%s", order_id) continue - if item.trade_date and item.trade_date != today and item.phase in {DONE, READY}: - item.trade_date, item.phase = "", READY + if sum(row.volume for row in rows) != pending.qty: + continue + if any(row.status not in TERMINAL_STATUSES for row in rows): + continue + if any(row.status == "56" and row.traded_volume != row.volume for row in rows): + continue + qty = sum(row.traded_volume for row in rows) + amounts = [row.trade_amount if row.trade_amount > 0 else row.trade_price * row.traded_volume + for row in rows if row.traded_volume > 0] + if any(not math.isfinite(amount) or amount <= 0 for amount in amounts): + continue + amount = sum(amounts) + cost = amount / qty if qty else 0.0 + item = self.items.setdefault(pending.code, TStateItem(pending.code)) + if pending.kind == "base": + item.base_order_id = order_id + item.base_qty, item.base_cost = qty, cost + elif pending.kind == "sell": + item.trade_date = today # 跨日成交也占用确认当天的一轮。 + item.sell_order_id = order_id + item.sell_qty, item.sell_price = qty, cost + item.buy_qty, item.buy_cost = 0, 0.0 + item.phase = SOLD if qty else READY + else: + total = item.buy_qty + qty + item.buy_cost = (item.buy_qty * item.buy_cost + amount) / total if total else 0.0 + item.buy_qty = total + item.buy_order_id = order_id + item.phase = DONE if total >= item.sell_qty else SOLD + if item.phase == DONE: + item.trade_date = today + # 零成交撤单也记录，保留计划、实际数量、均价和柜台终态。 + self.records.append({**asdict(pending), "confirmed_date": today, + "filled_qty": qty, "filled_cost": cost, + "amount": amount, "statuses": [row.status for row in rows]}) + del self.pending[order_id] + + for position in positions: + code = position.stock_code + if position.volume <= 0 or self.busy(code): + continue + if code not in self.items and math.isfinite(position.open_price) and position.open_price > 0: + self.items[code] = TStateItem(code, position.volume, position.open_price) + self.records.append({"kind": "import", "code": code, "date": today, + "filled_qty": position.volume, "filled_cost": position.open_price}) + log.warning("[ZT 底仓] 首次接管 %s，使用当前均价，无法还原历史成本", code) + + for item in self.items.values(): + # 未买回的轮次跨日继续，不删除零持仓的做 T 债务。 + if item.trade_date != today and item.phase == DONE and not self.busy(item.code): + item.phase, item.trade_date = READY, "" + item.sell_qty = item.buy_qty = 0 + item.sell_price = item.buy_cost = 0.0 item.sell_order_id = item.buy_order_id = "" - item.sell_qty = 0 - item.sell_price = 0.0 - if item.phase == SELLING and _completed(by_local_id.get(item.sell_order_id)): - item.phase = SOLD - elif item.phase == BUYING and _completed(by_local_id.get(item.buy_order_id)): - item.phase = DONE - self.set(item) self.save() def save(self) -> None: - with self.lock: - self.path.parent.mkdir(parents=True, exist_ok=True) - temporary = self.path.with_suffix(self.path.suffix + ".tmp") - temporary.write_text(json.dumps({key: asdict(value) for key, value in self.items.items()}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") - temporary.replace(self.path) + self.path.parent.mkdir(parents=True, exist_ok=True) + temporary = self.path.with_suffix(self.path.suffix + ".tmp") + payload = {"items": {key: asdict(item) for key, item in self.items.items()}, + "pending": {key: asdict(item) for key, item in self.pending.items()}, + "records": self.records} + temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8") + temporary.replace(self.path) - def _load(self) -> dict[str, TStateItem]: - try: - raw = json.loads(self.path.read_text(encoding="utf-8")) - except FileNotFoundError: - return {} - except (OSError, json.JSONDecodeError) as exc: - raise ValueError(f"[ZT 状态] 读取失败: {exc}") from exc - if not isinstance(raw, dict): - raise ValueError("[ZT 状态] 根节点必须是对象") - return {code: TStateItem(**value) for code, value in raw.items()} - - -def _completed(orders: list[OrderItem] | None) -> bool: - return bool(orders) and all(order.status == "56" for order in orders) + def _load(self) -> None: + if not self.path.is_file(): + return + raw = json.loads(self.path.read_text(encoding="utf-8")) + # 兼容原 ZT 文件，保留原底仓成本；没有额外版本字段。 + self.items = {code: TStateItem(**item) for code, item in raw.get("items", raw).items()} + self.pending = {key: PendingOrder(**item) for key, item in raw.get("pending", {}).items()} + self.records = raw.get("records", []) + if "items" not in raw: + # 旧记录只有提交行情价，不把它伪装成真实成交历史。 + self.records.append({"kind": "legacy_import", "items": raw}) + for item in self.items.values(): + if item.phase in {"SELLING", "BUYING", SOLD}: + raise ValueError(f"[ZT 状态] {item.code} 旧做 T 轮次未结束，需先核对成交再迁移") diff --git a/py-client/strategy/zt/watch.py b/py-client/strategy/zt/watch.py new file mode 100644 index 0000000..b5f3626 --- /dev/null +++ b/py-client/strategy/zt/watch.py @@ -0,0 +1,104 @@ +import logging as log +from dataclasses import dataclass +from datetime import datetime, timedelta +from threading import Lock + + +@dataclass(slots=True) +class _Entry: + last_close: float + expires_at: datetime + + +class DipWatch: + """观察价格低点，并在价格达到指定反弹幅度时触发。""" + + def __init__( + self, + expire_seconds: float = 300, + rebound_threshold: float = 1.5, # 反弹力度 1.5% + ) -> None: + self.expire_seconds = expire_seconds + self.rebound_threshold = rebound_threshold + self.data: dict[str, _Entry] = {} + self.lock = Lock() + + def triggered( + self, + tag: str, + code: str, + price: float, + now: datetime | None = None, + ) -> bool: + """更新观察价格；达到反弹阈值时返回 ``True``。""" + if price <= 0: + log.warning("[%s Watch] %s 价格无效：%.2f", tag, code, price) + return False + + current = now or datetime.now() + with self.lock: + watch = self.data.get(code) + + if watch is None: + self._start(code, price, current) + log.info( + "[%s Watch] %s 开始观察，收盘价=%.2f", + tag, + code, + price, + ) + return False + + if current >= watch.expires_at: + self._start(code, price, current) + log.info("[%sWatch] %s 观察已过期，重新观察，收盘价=%.2f", tag, code, price) + return False + + if price < watch.last_close: + old_price = watch.last_close + self._start(code, price, current) + log.info( + "[%s Watch] %s 刷新低点，原收盘价=%.2f，新收盘价=%.2f", + tag, + code, + old_price, + price, + ) + return False + + rebound = (price - watch.last_close) / watch.last_close * 100 + if rebound < self.rebound_threshold: + log.info( + "[%s Watch] %s 等待反弹，收盘价=%.2f，现价=%.2f，反弹=%.2f%%，阈值=%.2f%%", + tag, + code, + watch.last_close, + price, + rebound, + self.rebound_threshold, + ) + return False + + del self.data[code] + log.info( + "[%s Watch] %s 反弹触发，收盘价=%.2f，现价=%.2f，反弹=%.2f%%", + tag, + code, + watch.last_close, + price, + rebound, + ) + return True + + def forget(self, code: str) -> None: + """清除指定股票的价格观察状态。""" + with self.lock: + removed = self.data.pop(code, None) + if removed is not None: + log.info("[Watch] %s 已清除观察状态", code) + + def _start(self, code: str, price: float, now: datetime) -> None: + self.data[code] = _Entry( + last_close=price, + expires_at=now + timedelta(seconds=self.expire_seconds), + )