diff --git a/py-client/libs/state.py b/py-client/libs/state.py index 58b890a..89e82b2 100644 --- a/py-client/libs/state.py +++ b/py-client/libs/state.py @@ -42,7 +42,7 @@ CREATE TABLE IF NOT EXISTS deals ( ref INTEGER NOT NULL DEFAULT 0, order_ref TEXT NOT NULL DEFAULT '', direction INTEGER NOT NULL DEFAULT 0, - offset_flag INTEGER NOT NULL CHECK (offset_flag IN (23, 24, 48, 49)), + offset_flag INTEGER NOT NULL CHECK (offset_flag IN (48, 49)), price REAL NOT NULL CHECK (price >= 0), volume INTEGER NOT NULL CHECK (volume > 0), trade_amount REAL NOT NULL CHECK (trade_amount > 0), @@ -58,6 +58,14 @@ CREATE INDEX IF NOT EXISTS idx_deals_stock_code_date ON deals (stock_code); CREATE INDEX IF NOT EXISTS idx_deals_date_time ON deals (trade_date); """ +DELAS_INSERT_SQL = """ + INSERT INTO deals ( + stock_code, order_sys_id, order_local_id, ref, order_ref, + direction, offset_flag, price, volume, trade_amount, + trade_date, trade_time, remark, close_profit, is_arch + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0) +""" + @dataclass(slots=True) class StateItem: """策略状态字段；同步账户底仓时无法获知的委托编号留空。""" @@ -83,6 +91,7 @@ _UPSERT_STATE = ( ) + class State: """单写入者使用的 SQLite 存储；公开缓存仅在事务成功后替换。 @@ -144,53 +153,44 @@ class State: deals = self._read_deals(db) self.deals, self.deals_sys_ids = deals, set(deals) - @staticmethod - def _insert_deals(db: sqlite3.Connection, deals: list[DealItem], - existing_ids: set[str] | None = None) -> None: - """共享事务内保存成交；空本地编号使用明确的待核对标记。""" - existing = existing_ids if existing_ids is not None else { - row[0] for row in db.execute('SELECT order_sys_id FROM deals')} + def sync_deals(self, deals: list[DealItem]) -> None: + """按 order_sys_id 只追加成交时间超过 30 秒的新记录，保留全部历史记录。""" + now = datetime.now() new_deals: dict[str, DealItem] = {} for deal in deals: - if deal.order_sys_id not in existing: - new_deals.setdefault(deal.order_sys_id, deal) + if deal.order_sys_id in self.deals_sys_ids: + continue + date = (deal.trade_date or now.date().isoformat()).replace('-', '') + time = deal.trade_time.replace(':', '') + traded_at = datetime.strptime(f'{date} {time}', '%Y%m%d %H%M%S') + if (now - traded_at).total_seconds() <= 30: + continue + new_deals.setdefault(deal.order_sys_id, deal) if not new_deals: return - today = datetime.now().date().isoformat() - values = [] - for deal in new_deals.values(): - amount = deal.trade_amount if deal.trade_amount > 0 else deal.price * deal.volume - if not math.isfinite(deal.price) or not math.isfinite(amount): - raise ValueError('Trade price and amount must be finite') - date = deal.trade_date or today - if len(date) == 8 and date.isdigit(): - date = f'{date[:4]}-{date[4:6]}-{date[6:]}' - values.append(( - deal.stock_code, deal.order_sys_id, - deal.get_local_order_id.strip() or f'{UNATTRIBUTED_PREFIX}{deal.order_sys_id}', - deal.ref, deal.order_ref, deal.direction, deal.offset_flag, - deal.price, deal.volume, amount, date, deal.trade_time, - deal.remark, deal.close_profit, - )) - # 唯一键负责最终去重，避免旧缓存造成重复插入失败。 - db.executemany(""" - INSERT INTO deals ( - stock_code, order_sys_id, order_local_id, ref, order_ref, - direction, offset_flag, price, volume, trade_amount, - trade_date, trade_time, remark, close_profit, is_arch - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0) - ON CONFLICT(order_sys_id) DO NOTHING - """, values) - def sync_deals(self, deals: list[DealItem]) -> None: - """按成交编号只追加；同批保留首笔，任一无效成交回滚整批。""" - if all(deal.order_sys_id in self.deals_sys_ids for deal in deals): - return + # 插入记录。 with closing(self._connect()) as db, db: - db.execute('BEGIN IMMEDIATE') - self._insert_deals(db, deals) - cached_deals = self._read_deals(db) - self.deals, self.deals_sys_ids = cached_deals, set(cached_deals) + db.execute('BEGIN') + today = now.date().isoformat() + values = [] + for deal in new_deals.values(): + amount = deal.trade_amount if deal.trade_amount > 0 else deal.price * deal.volume + if not math.isfinite(deal.price) or not math.isfinite(amount): + raise ValueError('Trade price and amount must be finite') + date = deal.trade_date or today + if len(date) == 8 and date.isdigit(): + date = f'{date[:4]}-{date[4:6]}-{date[6:]}' + values.append(( + deal.stock_code, deal.order_sys_id, + deal.get_local_order_id.strip(), + deal.ref, deal.order_ref, deal.direction, deal.offset_flag, + deal.price, deal.volume, amount, date, deal.trade_time, + deal.remark, deal.close_profit, + )) + db.executemany(DELAS_INSERT_SQL, values) + cache_deals = self._read_deals(db) + self.deals, self.deals_sys_ids = cache_deals, set(cache_deals) def sync_state(self, positions: list[PositionItem]) -> None: """同步完整持仓：新增底仓、保留已有状态、删除已清仓证券。 @@ -221,172 +221,83 @@ class State: state = self._read_state(db) self.state = state - @staticmethod - def _archive_stock(db: sqlite3.Connection, code: str, snapshot_dedup: bool = True) -> None: - """在调用方的保存点内完成一只证券的记账及成交标记。""" - current = db.execute('SELECT * FROM state WHERE stock_code = ?', (code,)).fetchone() - state = dict(current) if current else asdict(StateItem(stock_code=code)) - pending = db.execute(""" - SELECT * FROM deals WHERE stock_code = ? AND is_arch = 0 - ORDER BY trade_date, REPLACE(trade_time, ':', ''), id - """, (code,)).fetchall() - if any(deal['order_local_id'].startswith(UNATTRIBUTED_PREFIX) for deal in pending): - raise ValueError('Unattributed trade requires reconciliation') - total_qty = state['base_qty'] + state['added_qty'] - if ( - snapshot_dedup and total_qty > 0 - and all(deal['offset_flag'] == FLAG_BUY for deal in pending) - and sum(deal['volume'] for deal in pending) == total_qty - ): - # 按快照去重规则，仅标记这批成交，原仓位数量和成本均保留。 - # 卖出和买卖混合批次不能用此规则，否则会漏掉真实减仓。 - db.executemany( - 'UPDATE deals SET is_arch = 1 WHERE id = ? AND is_arch = 0', - [(deal['id'],) for deal in pending], - ) - return - for deal in pending: - if deal['offset_flag'] == FLAG_BUY: - bucket = 'base' if deal['order_local_id'].startswith('zt-base-') else 'added' - total = state[f'{bucket}_qty'] + deal['volume'] - state[f'{bucket}_price'] = ( - state[f'{bucket}_qty'] * state[f'{bucket}_price'] + deal['trade_amount'] - ) / total - state[f'{bucket}_qty'] = total - state[f'{bucket}_order_local_id'] = deal['order_local_id'] - state[f'{bucket}_created_at'] = f"{deal['trade_date']} {deal['trade_time']}".strip() - elif deal['offset_flag'] == FLAG_SELL: - qty = deal['volume'] - total = state['base_qty'] + state['added_qty'] - if qty > total: - raise ValueError(f'Sell volume {qty} exceeds recorded holdings {total}') - if qty < state['added_qty']: - state['added_qty'] -= qty - else: - state['base_qty'] = total - qty - state['added_qty'] = 0 - state['added_price'] = 0.0 - state['added_order_local_id'] = state['added_created_at'] = '' - if state['base_qty'] == 0: - state.update(asdict(StateItem(stock_code=code))) - else: - raise ValueError(f"Unsupported trade flag: {deal['offset_flag']}") - if state['base_qty'] + state['added_qty'] == 0: - db.execute('DELETE FROM state WHERE stock_code = ?', (code,)) - else: - # 保留原记录主键及策略状态，包括同批清仓后重新建仓。 - if current: - state['status'] = current['status'] - db.execute(_UPSERT_STATE, state) - db.execute( - 'UPDATE deals SET is_arch = 1 WHERE stock_code = ? ' - 'AND is_arch = 0', (code,), - ) + def merge_deals(self) -> dict[str, dict]: + """按本地委托编号汇总未归档成交，返回新字典，不修改原始记录。 - def _archive_pending(self, db: sqlite3.Connection, snapshot_dedup: bool = True, - blocked_codes: set[str] | None = None) -> None: - """保留现有逐证券保存点；ZT 增量路径禁用数量相等推断。""" - codes = db.execute( - 'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0' - ).fetchall() - for row in codes: - code = row['stock_code'] - if blocked_codes and code in blocked_codes: + 数量、金额和平仓盈亏累加，价格为总金额除以总数量； + 其余字段（包括成交编号和时间）保留同组首笔记录的值。 + """ + merged: dict[str, dict] = {} + for deal in self.deals.values(): + if deal['is_arch'] != 0: continue - db.execute('SAVEPOINT archive_stock') - try: - self._archive_stock(db, code, snapshot_dedup) - except (ValueError, sqlite3.IntegrityError) as exc: - db.execute('ROLLBACK TO archive_stock') - log.warning('[归档] %s 失败，保留未归档成交：%s', code, exc) - finally: - db.execute('RELEASE archive_stock') + order_local_id = deal['order_local_id'] + if order_local_id not in merged: + merged[order_local_id] = deal.copy() + else: + item = merged[order_local_id] + item['stock_code']=deal['stock_code'] + item['volume'] += deal['volume'] + item['trade_amount'] += deal['trade_amount'] + item['close_profit'] += deal['close_profit'] + for item in merged.values(): + item['price'] = item['trade_amount'] / item['volume'] + return merged def archiving(self) -> None: - """归档未处理成交；单证券失败回滚并保留重试，不影响其他证券。""" + """先合并缓存中的未归档成交再计算；单证券失败回滚并保留重试。""" + merged = self.merge_deals() + if not merged: + return + with closing(self._connect()) as db, db: db.execute('BEGIN IMMEDIATE') - self._archive_pending(db) + for order_local_id, deal in merged.items(): + db.execute('SAVEPOINT archive_stock') + code = deal['stock_code'] + try: + result = db.execute('SELECT * FROM state WHERE stock_code = ?', (code,)).fetchone() + state = dict(result) if result else asdict(StateItem(stock_code=code)) + + if deal['offset_flag'] == FLAG_BUY: + bucket = 'base' if deal['order_local_id'].startswith('zt-base-') else 'added' + state[f'{bucket}_price'] = deal['price'] + state[f'{bucket}_qty'] = deal['volume'] + state[f'{bucket}_order_local_id'] = deal['order_local_id'] + state[f'{bucket}_created_at'] = f"{deal['trade_date']} {deal['trade_time']}".strip() + db.execute(_UPSERT_STATE, state) + elif deal['offset_flag'] == FLAG_SELL: + newState = StateItem(stock_code=code) + qty = deal['volume'] + total = state['base_qty'] + state['added_qty'] + if qty > total: + raise ValueError(f'Sell volume {qty} exceeds recorded holdings {total}') + elif qty == total: + # 清仓底仓与补仓 + newState.status='CLEAR' + elif qty == state['added_qty']: + # 清仓补仓 + newState.base_qty = state['base_qty'] + newState.base_price = state['base_price'] + newState.base_order_local_id = state['base_order_local_id'] + newState.base_created_at = state['base_created_at'] + elif qty == state['base_qty']: + # 清仓底仓 + newState.status='CLEAR' + else: + raise ValueError(f'Sell volume {qty} holdings {total}') + + if newState.status == 'CLEAR': + db.execute('DELETE FROM state WHERE stock_code = ?', (code,)) + else: + # 保留原记录主键及策略状态，包括同批清仓后重新建仓。 + db.execute(_UPSERT_STATE, asdict(newState)) + db.execute('UPDATE deals SET is_arch = 1 WHERE order_local_id = ? ''AND is_arch = 0', (order_local_id,)) + except (ValueError, sqlite3.IntegrityError) as exc: + db.execute('ROLLBACK TO archive_stock') + log.warning('[归档] %s 失败，保留未归档成交：%s', code, exc) + finally: + db.execute('RELEASE archive_stock') state = self._read_state(db) deals = self._read_deals(db) self.state, self.deals, self.deals_sys_ids = state, deals, set(deals) - - def sync_account(self, positions: list[PositionItem], deals: list[DealItem], *, initialize: bool = False) -> None: - """ZT 原子同步。初始化使用已核对快照；恢复只应用增量、不覆盖库存。 - - 空库存且无成交历史时允许初始化。快照缺失或 - 未归档成交只隔离对应证券，后续成交到齐并核对一致后自动恢复。 - """ - holdings = {p.stock_code: p for p in positions if p.volume > 0} - with closing(self._connect()) as db, db: - db.execute('BEGIN IMMEDIATE') - db.execute('''CREATE TABLE IF NOT EXISTS zt_rejected_deals ( - stock_code TEXT NOT NULL, order_sys_id TEXT NOT NULL, - payload TEXT NOT NULL, error TEXT NOT NULL, - PRIMARY KEY (stock_code, order_sys_id) - )''') - if initialize: - if (db.execute('SELECT 1 FROM state LIMIT 1').fetchone() - or db.execute('SELECT 1 FROM deals LIMIT 1').fetchone()): - raise ValueError('ZT state already initialized; cannot overwrite existing data') - created_at = datetime.now().isoformat(timespec='seconds') - for code, position in holdings.items(): - if not code or not math.isfinite(position.open_price) or position.open_price <= 0: - raise ValueError('Initial position code and cost must be valid') - db.execute(_UPSERT_STATE, asdict(StateItem( - stock_code=code, base_qty=position.volume, - base_price=position.open_price, base_created_at=created_at, - ))) - self._insert_deals(db, deals) - # 这些成交已包含在已核对的基准中，不再增减快照库存。 - db.execute('UPDATE deals SET is_arch = 1') - else: - self._insert_account_deals(db, deals) - rejected = {row[0] for row in db.execute('SELECT stock_code FROM zt_rejected_deals')} - self._archive_pending(db, snapshot_dedup=False, blocked_codes=rejected) - state = self._read_state(db) - cached_deals = self._read_deals(db) - blocked = {d['stock_code'] for d in cached_deals.values() if d['is_arch'] != 1} - blocked.update(row[0] for row in db.execute('SELECT stock_code FROM zt_rejected_deals')) - for code in set(state) | set(holdings): - row = state.get(code, {}) - recorded = row.get('base_qty', 0) + row.get('added_qty', 0) - actual = holdings[code].volume if code in holdings else 0 - if recorded != actual: - blocked.add(code) - self.state, self.deals, self.deals_sys_ids = state, cached_deals, set(cached_deals) - self.blocked_codes = blocked - if blocked: - log.warning('[ZT 同步] 以下证券状态待核对，暂停交易：%s', ', '.join(sorted(blocked))) - - def _insert_account_deals(self, db: sqlite3.Connection, deals: list[DealItem]) -> None: - """仅 ZT 增量路径隔离坏成交；保留原始字段，缺席回报不会解除隔离。""" - existing = {row[0] for row in db.execute('SELECT order_sys_id FROM deals')} - for deal in deals: - if not deal.stock_code: - raise ValueError('Trade without stock code cannot be isolated') - db.execute('SAVEPOINT insert_zt_deal') - try: - if not deal.stock_code or not deal.order_sys_id: - raise ValueError('Trade code and order ID are required') - if (deal.offset_flag not in (FLAG_BUY, FLAG_SELL) - or not math.isfinite(deal.volume) or deal.volume <= 0 - or int(deal.volume) != deal.volume - or not math.isfinite(deal.price) or deal.price < 0 - or not math.isfinite(deal.trade_amount) - or not math.isfinite(deal.close_profit)): - raise ValueError('Invalid trade direction, volume or amount') - self._insert_deals(db, [deal], existing) - except (ValueError, TypeError, OverflowError, sqlite3.IntegrityError) as exc: - db.execute('ROLLBACK TO insert_zt_deal') - db.execute('''INSERT OR REPLACE INTO zt_rejected_deals - (stock_code, order_sys_id, payload, error) VALUES (?, ?, ?, ?)''', - (deal.stock_code, deal.order_sys_id, json.dumps(asdict(deal), ensure_ascii=False), str(exc))) - log.warning('[ZT 成交] 隔离证券=%s，委托=%s，原因=%s', - deal.stock_code, deal.order_sys_id, exc) - else: - existing.add(deal.order_sys_id) - db.execute('DELETE FROM zt_rejected_deals WHERE stock_code = ? AND order_sys_id = ?', - (deal.stock_code, deal.order_sys_id)) - finally: - db.execute('RELEASE insert_zt_deal') diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index 029617d..c8c2af4 100644 --- a/py-client/strategy/zt/boot.py +++ b/py-client/strategy/zt/boot.py @@ -15,7 +15,7 @@ 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 +from sdk import Client, DealItem, PositionItem from .open import open_signal from .positions import manage_positions from libs.snapshot import cache_portfolio @@ -55,7 +55,7 @@ def StartZT() -> None: log.info('[启动] ZT策略已启动，账户=%s，信号=%d，持仓=%d', config.account_config.account_id, len(signals), len(positions)) cache_portfolio(config.account_config.account_id, assets, positions, deals) - state.sync_account(positions, deals, initialize=initialize) + _sync_state(state, positions, deals, initialize=initialize) run.profit_tracker.sync_positions(positions, state) run.orders.refresh(client, portfolio.orders, cancel_prefix='zt-') Overview(assets, positions, config.account_config) @@ -94,6 +94,26 @@ def StartZT() -> None: client.close() +def _sync_state(state: State, positions: list[PositionItem], deals: list[DealItem], + *, initialize: bool = False) -> None: + """使用 State 的独立接口同步，交易前核对归档结果与账户持仓。""" + if initialize: + state.sync_state(positions) + state.sync_deals(deals) + state.archiving() + + holdings = {p.stock_code: p.volume for p in positions if p.volume > 0} + blocked = {d.stock_code for d in deals if d.order_sys_id not in state.deals_sys_ids} + blocked.update(d['stock_code'] for d in state.deals.values() if d['is_arch'] != 1) + for code in state.state.keys() | holdings.keys(): + row = state.get_by_code(code) + if row.get('base_qty', 0) + row.get('added_qty', 0) != holdings.get(code, 0): + blocked.add(code) + state.blocked_codes = blocked + if blocked: + log.warning('[ZT 同步] 状态待核对，暂停交易：%s', ', '.join(sorted(blocked))) + + def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None: now = datetime.now() if not trading_time(now): @@ -114,7 +134,7 @@ def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None: position_codes = list(portfolio.positions) cache_portfolio(run.account_cfg.account_id, assets, positions, deals) - state.sync_account(positions, deals) + _sync_state(state, positions, deals) run.profit_tracker.sync_positions(positions, state) run.orders.refresh(run.client, portfolio.orders, cancel_prefix='zt-') except Exception: