"""ZT 启动与串行调度:成交同步、买回、卖出、建仓。""" import logging as log import time from contextlib import closing from datetime import datetime from pathlib import Path from tempfile import TemporaryDirectory import config from libs.calc import trading_time from libs.collector import collector_push from libs.grid_take_profit import GridTrailingTracker from libs.market import market_allow_open from libs.order import OrderBook from libs.overview import Overview from libs.runtime import Runtime from libs.signal import SignalItem, init_signals from libs.state import State from libs.watch import DipWatch from sdk import Client, DealItem, PositionItem from .open import open_signal from .positions import manage_positions, t_rounds def StartZT() -> None: with Client(config.global_config.qmt_base_url, config.global_config.qmt_token, config.HTTP_TIMEOUT) as client: state = State(Path(config.global_config.qmt_data_dir) / f'zt_{config.account_config.account_id}_state.db') run = Runtime( client=client, global_cfg=config.global_config, account_cfg=config.account_config, orders=OrderBook('zt'), open_watch=DipWatch(), add_watch=DipWatch(), profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct), ) # 先读取成交,再读取持仓,减少成交已入账而快照仍未更新的情况。 deals = client.deals() portfolio = client.portfolio() positions = list(portfolio.positions.values()) sync_account_state(state, positions, deals, initialize=not state.state and not state.deals) run.orders.refresh(client, portfolio.orders) signals = init_signals(config.global_config, ['dcm']) Overview(portfolio.assets, positions, config.account_config) log.info('[ZT] 启动,账户=%s,信号=%d', config.account_config.account_id, len(signals)) while datetime.now().hour < 15: try: RunOnce(run, state, signals) except Exception: log.exception('[ZT] 本轮失败,下一轮重试') time.sleep(30 - time.time() % 30) # 收盘后补记最后一轮成交,不再下单。 deals = client.deals() sync_account_state(state, list(client.portfolio().positions.values()), deals) def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None: now = datetime.now() if not trading_time(now): return deals = run.client.deals() portfolio = run.client.portfolio() assets = portfolio.assets positions = list(portfolio.positions.values()) run.orders.refresh(run.client, portfolio.orders) sync_account_state(state, positions, deals) # 收盘集合竞价前停止提交新委托,继续保存成交。 if (now.hour, now.minute) >= (14, 57): return rounds = t_rounds(state) pending = {code for code, item in rounds.items() if item['sold'] > item['bought']} candidates = {s.code: s for s in signals if s.code not in portfolio.positions and s.code not in state.state and s.code not in pending} codes = list(dict.fromkeys(list(state.state) + sorted(pending) + list(candidates))) ticks = run.client.full_tick(codes) if codes else {} force = (now.hour, now.minute) >= (14, 50) available = manage_positions(run, state, ticks, positions, rounds, assets.available, now.date().isoformat(), force) # 尚未买回时不分走资金;买回与新建仓使用同一份剩余资金。 if not force and not pending and available >= assets.total * run.account_cfg.min_cash_ratio: if candidates and market_allow_open(): budget = max(0.0, available - assets.total * run.account_cfg.min_cash_ratio) open_signal(run, ticks, list(candidates.values()), budget) try: collector_push(run.account_cfg.account_id, assets, positions) except Exception: log.exception('[ZT] 持仓上报失败') def sync_account_state( state: State, positions: list[PositionItem], deals: list[DealItem], *, initialize: bool = False, ) -> None: """初次持仓作为底仓;后续只按成交减仓,避免延迟快照删除持仓。""" zt_deals = [d for d in deals if d.get_local_order_id.startswith('zt-')] if initialize: if state.state or state.deals: raise ValueError('ZT 初始化仅允许空状态库,禁止覆盖已有持仓和成交') # 在同目录临时库完成初始化,再原子替换空库,避免中途退出留下半份快照。 with TemporaryDirectory(dir=state.path.parent, prefix='zt-init-') as directory: initial = State(Path(directory) / 'state.db') initial.sync_deals(zt_deals) initial.sync_state(positions) # 初始快照已包含历史成交;保留成交计算做 T 欠仓,但不再累加持仓。 with closing(initial._connect()) as db, db: db.execute('UPDATE deals SET is_arch = 1') initial.path.replace(state.path) state.load() return state.sync_deals(zt_deals) # 在外层统一归档:新增成交写入后处理,也重试此前失败的未归档成交。 state.archiving() # 从刷新后的成交缓存检查失败记录,避免归档未完成时继续交易。 pending = sorted({d['stock_code'] for d in state.deals.values() if d['is_arch'] == 0}) if pending: raise ValueError(f'ZT 成交归档未完成:{pending},原因见归档日志') traded = {d['stock_code'] for d in state.deals.values()} # sync_state 要求完整持仓。保留成交账本中的现有仓位,避免延迟快照删仓; # 只从账户快照补入没有策略成交历史的股票,避免刚卖完又被旧快照重建。 holdings = [PositionItem(stock_code=code, volume=row['base_qty'] + row['added_qty']) for code, row in state.state.items()] holdings.extend(p for p in positions if p.stock_code not in traded and p.stock_code not in state.state) state.sync_state(holdings)