fix zt

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 sync_deals(self, deals: list[DealItem]) -> None:
-    def _insert_deals(db: sqlite3.Connection, deals: list[DealItem],
+        """按 order_sys_id 只追加成交时间超过 30 秒的新记录，保留全部历史记录。"""
-                      existing_ids: set[str] | None = None) -> None:
+        now = datetime.now()
-        """共享事务内保存成交；空本地编号使用明确的待核对标记。"""
-        existing = existing_ids if existing_ids is not None else {
-            row[0] for row in db.execute('SELECT order_sys_id FROM deals')}
         new_deals: dict[str, DealItem] = {}
         for deal in deals:
-            if deal.order_sys_id not in existing:
+            if deal.order_sys_id in self.deals_sys_ids:
-                new_deals.setdefault(deal.order_sys_id, deal)
+                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')
+            db.execute('BEGIN')
-            self._insert_deals(db, deals)
+            today = now.date().isoformat()
-            cached_deals = self._read_deals(db)
+            values = []
-        self.deals, self.deals_sys_ids = cached_deals, set(cached_deals)
+            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 merge_deals(self) -> dict[str, dict]:
-    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 _archive_pending(self, db: sqlite3.Connection, snapshot_dedup: bool = True,
+        数量、金额和平仓盈亏累加，价格为总金额除以总数量；
-                         blocked_codes: set[str] | None = None) -> None:
+        其余字段（包括成交编号和时间）保留同组首笔记录的值。
-        """保留现有逐证券保存点；ZT 增量路径禁用数量相等推断。"""
+        """
-        codes = db.execute(
+        merged: dict[str, dict] = {}
-            'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0'
+        for deal in self.deals.values():
-        ).fetchall()
+            if deal['is_arch'] != 0:
-        for row in codes:
-            code = row['stock_code']
-            if blocked_codes and code in blocked_codes:
                 continue
-            db.execute('SAVEPOINT archive_stock')
+            order_local_id = deal['order_local_id']
-            try:
+            if order_local_id not in merged:
-                self._archive_stock(db, code, snapshot_dedup)
+                merged[order_local_id] = deal.copy()
-            except (ValueError, sqlite3.IntegrityError) as exc:
+            else:
-                db.execute('ROLLBACK TO archive_stock')
+                item = merged[order_local_id]
-                log.warning('[归档] %s 失败，保留未归档成交：%s', code, exc)
+                item['stock_code']=deal['stock_code']
-            finally:
+                item['volume'] += deal['volume']
-                db.execute('RELEASE archive_stock')
+                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')

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: