From 6e554e15658b0a8be362c19ecdbf6d797bc70be6 Mon Sep 17 00:00:00 2001 From: yanweidong Date: Fri, 11 Sep 2026 10:44:34 +0800 Subject: [PATCH] fix bug --- py-client/etc/_global.yaml | 2 +- py-client/libs/order.py | 1 + py-client/libs/state.py | 35 ++++++++++++++---- py-client/strategy/trend/positions.py | 4 +-- py-client/strategy/zt/boot.py | 36 +++++++++++++++---- py-client/strategy/zt/open.py | 2 +- py-client/strategy/zt/positions.py | 4 +-- py-client/tests/test_state_archiving.py | 47 +++++++++++++++++-------- py-client/tests/test_zt_state.py | 27 ++++++++++++++ py-client/tests/test_zt_trading.py | 18 +++++++++- 10 files changed, 140 insertions(+), 36 deletions(-) diff --git a/py-client/etc/_global.yaml b/py-client/etc/_global.yaml index c7fd5a0..39ca18a 100644 --- a/py-client/etc/_global.yaml +++ b/py-client/etc/_global.yaml @@ -19,4 +19,4 @@ signals: dcm: {url: /a/dcm_signal, timezone: "*", gt_last_price_is_open: false} morning: {url: /a/morning_signal, timezone: "9:30-10:30", gt_last_price_is_open: true} tail: {url: /a/tail_signal, timezone: "14:30-14:55", gt_last_price_is_open: false} - arbitrage: {url: /a/arbitrage_signal, timezone: "*", gt_last_price_is_open: false} + arbitrage: {url: /a/arbitrage_signal, timezone: "13:00-14:50", gt_last_price_is_open: false} diff --git a/py-client/libs/order.py b/py-client/libs/order.py index f32d639..5773542 100644 --- a/py-client/libs/order.py +++ b/py-client/libs/order.py @@ -35,6 +35,7 @@ class OrderBook: def __init__( self, order_prefix: str, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 30 ) -> None: + self.order_prefix = order_prefix self.lock_timeout_sec = max(1, lock_timeout_sec) self.cancel_timeout_sec = timedelta(seconds=cancel_timeout_sec) self.data: list[OrderItem] = [] diff --git a/py-client/libs/state.py b/py-client/libs/state.py index da2a9c1..522621f 100644 --- a/py-client/libs/state.py +++ b/py-client/libs/state.py @@ -137,6 +137,7 @@ class State: amount, date, deal.trade_time, deal.remark, deal.close_profit, 0), ) self.load_deals() + def sync_state(self, positions: list[PositionItem]) -> None: """同步完整持仓：无状态则插入底仓，已有则保留，清仓则删除。 """ @@ -165,20 +166,29 @@ class State: ) self.load_state() - def archiving(self, *, base_order_prefix: str = '') -> dict[str, str]: - """将未归档成交累加到当前底仓和加仓；失败证券保留记录供重试。""" - errors = {} + def archiving(self) -> None: + """将未归档成交累加到当前底仓和加仓；失败证券保留记录供重试。 + + 归档是把成交反映到持仓状态中，再标记为已处理，不会删除成交记录。 + 本地委托号以 zt-base- 开头的买入计入底仓，其他买入计入补仓。 + 不返回结果；失败原因记录到日志，对应成交保留未归档标记供重试。 + """ with closing(self._connect()) as db, db: + # 提前取得数据库写入锁，让持仓更新和成交标记在同一事务内完成。 db.execute('BEGIN IMMEDIATE') + # 只找尚未处理的成交：23、48 是买入，24、49 是卖出。 codes = db.execute( - 'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0 AND offset_flag IN (23, 24, 48, 49)' + 'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0' ).fetchall() for entry in codes: code = entry['stock_code'] + # 每只股票设一个回滚点；这只处理失败时，不撤销其他股票的结果。 db.execute('SAVEPOINT archive_stock') try: current = db.execute('SELECT * FROM state WHERE stock_code = ?', (code,)).fetchone() + # 已有持仓就接着计算；没有记录则从底仓、补仓均为零开始。 state = dict(current) if current else asdict(StateItem(stock_code=code)) + # 按成交日期、时间、记录编号依次处理，保证先买后卖等顺序正确。 deals = db.execute( 'SELECT * FROM deals WHERE stock_code = ? AND is_arch = 0 AND offset_flag IN (23, 24, 48, 49) ' "ORDER BY trade_date, REPLACE(trade_time, ':', ''), id", (code,) @@ -186,34 +196,43 @@ class State: for deal in deals: qty = deal['volume'] if deal['offset_flag'] in (23, 48): - bucket = 'base' if base_order_prefix and deal['order_local_id'].startswith(base_order_prefix) else 'added' + # 按已有的 ZT 底仓委托号约定识别，无需调用方传入规则。 + bucket = 'base' if deal['order_local_id'].startswith('zt-base-') else 'added' total = state[f'{bucket}_qty'] + qty + # 新均价 =（原数量 × 原均价 + 本次成交金额）÷ 买入后总数量。 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() else: 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'] + state['added_qty'] == 0: + # 中途清仓先重置，后续若又买入，就从零重新累计。 state = asdict(StateItem(stock_code=code)) if state['base_qty'] + state['added_qty'] == 0: + # 全部成交处理完仍无持仓，则删除该股票的策略状态。 db.execute('DELETE FROM state WHERE stock_code = ?', (code,)) else: # 更新持仓，保留策略状态及已有记录主键。 state['status'] = current['status'] if current else state['status'] state.pop('id', None) columns = tuple(state) + # 没有该股票就新增，已有则更新仓位字段，不替换原记录主键。 db.execute( f"INSERT INTO state ({', '.join(columns)}) " f"VALUES ({', '.join(':' + key for key in columns)}) " @@ -221,17 +240,19 @@ class State: + ', '.join(f'{key} = excluded.{key}' for key in columns if key != 'stock_code'), state, ) + # 持仓处理成功后才标记成交，防止下次重复加仓或重复扣减。 db.execute( 'UPDATE deals SET is_arch = 1 WHERE stock_code = ? ' 'AND is_arch = 0 AND offset_flag IN (23, 24, 48, 49)', (code,) ) except (ValueError, sqlite3.IntegrityError) as exc: + # 撤销这只股票的全部归档修改，成交仍保持未归档，供下次重试。 db.execute('ROLLBACK TO archive_stock') - errors[code] = str(exc) log.warning('[归档] %s 失败，保留未归档成交：%s', code, exc) finally: + # 释放当前股票的回滚点；整个事务在退出外层 with 时提交。 db.execute('RELEASE archive_stock') + # 数据库提交完成后刷新内存缓存，让策略读到最新持仓和归档标记。 self.load() - return errors diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index aa57c0b..eaf2274 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -137,7 +137,7 @@ def handle_profit( volume = position.can_use_volume - position.can_use_volume % 100 if volume <= 0: return TradeDecision(False, "无可用整手持仓") - order_id = runtime.orders.new_order_id("SELL") + order_id = runtime.orders.new_order_id("TREN","SELL") request = PlaceOrderRequest( op=OP_SELL, code=position.stock_code, @@ -178,7 +178,7 @@ def handle_loss( if volume <= 0 or amount > available: return TradeDecision(False, "本轮可用资金不足") - order_id = runtime.orders.new_order_id("BUY") + order_id = runtime.orders.new_order_id("TREN","BUY") request = PlaceOrderRequest( op=OP_BUY, code=position.stock_code, diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index d491ade..68225ee 100644 --- a/py-client/strategy/zt/boot.py +++ b/py-client/strategy/zt/boot.py @@ -2,8 +2,10 @@ 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 @@ -45,7 +47,8 @@ def StartZT() -> None: log.exception('[ZT] 本轮失败，下一轮重试') time.sleep(30 - time.time() % 30) # 收盘后补记最后一轮成交，不再下单。 - sync_account_state(state, list(client.portfolio().positions.values()), client.deals()) + deals = client.deals() + sync_account_state(state, list(client.portfolio().positions.values()), deals) def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None: @@ -85,12 +88,31 @@ def sync_account_state( ) -> None: """初次持仓作为底仓；后续只按成交减仓，避免延迟快照删除持仓。""" zt_deals = [d for d in deals if d.get_local_order_id.startswith('zt-')] - state.sync_deals(zt_deals) if initialize: - state.sync_state(positions) + 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 - errors = state.archiving(base_order_prefix='zt-base-') - if errors: - raise ValueError(f'ZT 成交归档失败：{errors}') + 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()} - state.sync_state([p for p in positions if p.stock_code not in traded], remove_missing=False) + # 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) diff --git a/py-client/strategy/zt/open.py b/py-client/strategy/zt/open.py index 960d30e..b1baa57 100644 --- a/py-client/strategy/zt/open.py +++ b/py-client/strategy/zt/open.py @@ -42,7 +42,7 @@ def open_signal(run: Runtime, ticks, signals, available: float) -> float: # 等待价格从观察低点反弹，防止直接接下跌中的“飞刀”。 if not run.open_watch.triggered("ZT 建仓", item.code, price): continue - order_id = run.orders.new_order_id("ZT","base") + order_id = run.orders.new_order_id("zt", "base") request = PlaceOrderRequest( OP_BUY, item.code, volume, order_id, "zt" ) diff --git a/py-client/strategy/zt/positions.py b/py-client/strategy/zt/positions.py index 17cf87b..77758d0 100644 --- a/py-client/strategy/zt/positions.py +++ b/py-client/strategy/zt/positions.py @@ -74,7 +74,7 @@ def manage_positions( if not force_buy_back and not run.add_watch.triggered('ZT 买回', code, price): continue available -= amount - request = PlaceOrderRequest(OP_BUY, code, volume, run.orders.new_order_id('t-buy'), 'zt') + request = PlaceOrderRequest(OP_BUY, code, volume, run.orders.new_order_id('zt', 't-buy'), 'zt') if run.orders.place(run.client, request): run.add_watch.forget(code) log.info('[ZT 买回] %s %d 股%s', code, volume, '，尾盘买回' if force_buy_back else '') @@ -93,7 +93,7 @@ def manage_positions( volume = int(min(position.can_use_volume, recorded * run.account_cfg.zt_sell_ratio)) // 100 * 100 if volume < (200 if code.startswith('688') else 100): continue - request = PlaceOrderRequest(OP_SELL, code, volume, run.orders.new_order_id('t-sell'), 'zt') + request = PlaceOrderRequest(OP_SELL, code, volume, run.orders.new_order_id('zt', 't-sell'), 'zt') if run.orders.place(run.client, request): log.info('[ZT 卖出] %s %d 股，按实际成交买回', code, volume) except Exception: diff --git a/py-client/tests/test_state_archiving.py b/py-client/tests/test_state_archiving.py index 2dcaab7..17394e5 100644 --- a/py-client/tests/test_state_archiving.py +++ b/py-client/tests/test_state_archiving.py @@ -52,6 +52,20 @@ class ArchivingTests(unittest.TestCase): self.assertAlmostEqual(row['added_price'], 1620 / 150) self.assertEqual(row['added_order_local_id'], 'late') + def test_no_argument_archiving_recognizes_base_and_added_orders(self): + self.insert_deal('zt-base-first', 100, 1000, '10:00:00', flag=23) + self.insert_deal('zt-base-second', 100, 1200, '10:01:00', flag=48) + self.insert_deal('zt-t-buy-first', 100, 900, '10:02:00', flag=23) + self.assertIsNone(self.book.archiving()) + row = self.book.state['600000.SH'] + self.assertEqual((row['base_qty'], row['base_price']), (200, 11)) + self.assertEqual((row['added_qty'], row['added_price']), (100, 9)) + self.assertEqual(row['base_order_local_id'], 'zt-base-second') + self.assertTrue(all(deal['is_arch'] == 1 for deal in self.book.deals.values())) + saved = dict(row) + self.assertIsNone(self.book.archiving()) + self.assertEqual(self.book.state['600000.SH'], saved) + def test_sell_added_then_clear_base(self): self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) self.insert_deal('buy1', 40, 400, '09:59:00') @@ -88,8 +102,9 @@ class ArchivingTests(unittest.TestCase): def test_excess_sell_rolls_back(self): self.insert_deal('buy', 50, 500, '10:00:00') self.insert_deal('sell', 100, 1200, '10:01:00', flag=49) - errors = self.book.archiving() - self.assertIn('600000.SH', errors) + with self.assertLogs(level='WARNING') as logs: + self.assertIsNone(self.book.archiving()) + self.assertIn('600000.SH', '\n'.join(logs.output)) restarted = State(self.book.path) self.assertEqual(restarted.state, {}) self.assertTrue(all(deal['is_arch'] == 0 for deal in restarted.deals.values())) @@ -98,7 +113,7 @@ class ArchivingTests(unittest.TestCase): self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) self.insert_deal('buy', 100, 1000, '10:00:00', flag=23) self.insert_deal('sell', 50, 600, '10:01:00', flag=24) - self.assertEqual(self.book.archiving(), {}) + self.assertIsNone(self.book.archiving()) row = self.book.state['600000.SH'] self.assertEqual((row['base_qty'], row['added_qty']), (100, 50)) self.assertEqual(self.book.deals['buy']['offset_flag'], 23) @@ -113,8 +128,9 @@ class ArchivingTests(unittest.TestCase): db.execute("""CREATE TRIGGER fail_archive BEFORE UPDATE OF is_arch ON deals WHEN OLD.order_sys_id = 'second' BEGIN SELECT RAISE(ABORT, 'test failure'); END""") - errors = self.book.archiving() - self.assertIn('600001.SH', errors) + with self.assertLogs(level='WARNING') as logs: + self.assertIsNone(self.book.archiving()) + self.assertIn('600001.SH', '\n'.join(logs.output)) restarted = State(self.book.path) self.assertEqual(set(restarted.state), {'600000.SH'}) self.assertEqual(restarted.state, self.book.state) @@ -134,17 +150,17 @@ class ArchivingTests(unittest.TestCase): offset_flag=48, volume=100, price=8, trade_amount=800, trade_date='20260908', trade_time='100000', )]) - self.assertEqual(self.book.archiving(), {}) + self.assertIsNone(self.book.archiving()) row = self.book.state['600000.SH'] self.assertEqual((row['base_qty'], row['added_qty']), (100, 100)) self.assertEqual(self.book.deals['first']['is_arch'], 1) restarted = State(self.book.path) - self.assertEqual(restarted.archiving(), {}) + self.assertIsNone(restarted.archiving()) self.assertEqual(restarted.state, self.book.state) self.insert_deal('new_buy', 100, 1000, '10:01:00') with closing(self.book._connect()) as db, db: db.execute("UPDATE state SET status = 'CUSTOM' WHERE stock_code = '600000.SH'") - self.assertEqual(self.book.archiving(), {}) + self.assertIsNone(self.book.archiving()) row = self.book.state['600000.SH'] self.assertEqual((row['base_qty'], row['added_qty'], row['status']), (100, 200, 'CUSTOM')) @@ -153,8 +169,7 @@ class ArchivingTests(unittest.TestCase): with closing(self.book._connect()) as db, db: db.execute('UPDATE deals SET is_arch = 1') self.insert_deal('new', 50, 500, '10:01:00') - errors = self.book.archiving() - self.assertEqual(errors, {}) + self.assertIsNone(self.book.archiving()) self.assertEqual(self.book.state['600000.SH']['added_qty'], 50) self.assertEqual(self.book.deals['old']['is_arch'], 1) self.assertEqual(self.book.deals['new']['is_arch'], 1) @@ -166,7 +181,7 @@ class ArchivingTests(unittest.TestCase): self.assertEqual(self.book.state, {}) self.book = State(self.book.path) self.insert_deal('late', 100, 2000, '100100') - self.assertEqual(self.book.archiving(), {}) + self.assertIsNone(self.book.archiving()) row = self.book.state['600000.SH'] self.assertEqual(row['added_qty'], 100) self.assertEqual(row['added_price'], 20) @@ -174,21 +189,23 @@ class ArchivingTests(unittest.TestCase): def test_archive_sell_before_syncing_empty_positions(self): self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) self.insert_deal('sell', 100, 1000, '10:01:00', flag=49) - self.assertEqual(self.book.archiving(), {}) + self.assertIsNone(self.book.archiving()) self.book.sync_state([]) self.book = State(self.book.path) - self.assertEqual(self.book.archiving(), {}) + self.assertIsNone(self.book.archiving()) self.assertEqual(self.book.state, {}) self.assertEqual(self.book.deals['sell']['is_arch'], 1) def test_bad_stock_does_not_block_good_stock_and_can_retry(self): self.insert_deal('bad_sell', 100, 1500, '10:02:00', flag=49) self.insert_deal('good_buy', 100, 1000, '10:00:00', code='600001.SH') - self.assertIn('600000.SH', self.book.archiving()) + with self.assertLogs(level='WARNING') as logs: + self.assertIsNone(self.book.archiving()) + self.assertIn('600000.SH', '\n'.join(logs.output)) self.assertEqual(self.book.deals['good_buy']['is_arch'], 1) self.assertEqual(self.book.deals['bad_sell']['is_arch'], 0) self.insert_deal('late_buy', 100, 1000, '10:01:00') - self.assertEqual(self.book.archiving(), {}) + self.assertIsNone(self.book.archiving()) self.assertEqual(self.book.deals['bad_sell']['is_arch'], 1) self.assertNotIn('600000.SH', self.book.state) diff --git a/py-client/tests/test_zt_state.py b/py-client/tests/test_zt_state.py index c0411df..412e377 100644 --- a/py-client/tests/test_zt_state.py +++ b/py-client/tests/test_zt_state.py @@ -1,5 +1,6 @@ import tempfile import unittest +from unittest.mock import patch from pathlib import Path from libs.state import State @@ -49,6 +50,32 @@ class ZTStateTests(unittest.TestCase): self.assertEqual(self.state.state['600000.SH']['base_qty'], 100) self.assertEqual(self.state.deals['sell']['is_arch'], 0) + def test_failed_initialization_leaves_original_database_empty(self): + invalid = self.position(100) + invalid.open_price = float('inf') + with self.assertRaises(ValueError): + sync_account_state(self.state, [invalid], [self.deal('old', 100)], initialize=True) + restarted = State(self.state.path) + self.assertEqual((restarted.state, restarted.deals), ({}, {})) + sync_account_state(restarted, [self.position(100)], [self.deal('old', 100)], initialize=True) + self.assertEqual(restarted.deals['old']['is_arch'], 1) + + def test_initialization_cannot_overwrite_existing_database(self): + sync_account_state(self.state, [self.position(100)], [], initialize=True) + with self.assertRaises(ValueError): + sync_account_state(self.state, [], [], initialize=True) + self.assertEqual(State(self.state.path).state['600000.SH']['base_qty'], 100) + + def test_no_new_deals_still_retries_failed_archiving(self): + sync_account_state(self.state, [self.position(100)], [], initialize=True) + sell = self.deal('sell', 100, flag=24) + with patch.object(self.state, 'archiving'): + with self.assertRaises(ValueError): + sync_account_state(self.state, [], [sell]) + sync_account_state(self.state, [], [sell]) + self.assertEqual(self.state.state, {}) + self.assertEqual(self.state.deals['sell']['is_arch'], 1) + if __name__ == '__main__': unittest.main() diff --git a/py-client/tests/test_zt_trading.py b/py-client/tests/test_zt_trading.py index 2784510..5b03176 100644 --- a/py-client/tests/test_zt_trading.py +++ b/py-client/tests/test_zt_trading.py @@ -7,6 +7,7 @@ from unittest.mock import Mock, patch from config import AccountConfig from libs.grid_take_profit import GridState +from libs.order import OrderBook from libs.state import State from sdk import Assets, DealItem, PositionItem, Tick from strategy.zt import boot @@ -24,7 +25,7 @@ class ZTTradingTests(unittest.TestCase): self.run = SimpleNamespace(account_cfg=self.cfg, orders=Mock(), client=Mock(), profit_tracker=Mock(), add_watch=Mock(), open_watch=Mock()) self.run.orders.busy.return_value = False - self.run.orders.new_order_id.side_effect = lambda kind: f'zt-{kind}-order' + self.run.orders.new_order_id.side_effect = lambda prefix, kind: f'{prefix}-{kind}-order' self.run.profit_tracker.observe.return_value.state = GridState.RETREAT self.run.add_watch.triggered.return_value = True self.run.open_watch.triggered.return_value = True @@ -131,6 +132,21 @@ class ZTTradingTests(unittest.TestCase): [SimpleNamespace(code='688001.SH')], 2000) self.run.orders.place.assert_not_called() + def test_real_order_id_is_recognized_by_state_sync(self): + orders = OrderBook('zt') + self.run.orders.new_order_id.side_effect = orders.new_order_id + with patch('strategy.zt.open.datetime') as clock: + clock.now.return_value = datetime(2026, 9, 9, 10) + open_signal(self.run, {self.code: Tick(last_price=10)}, + [SimpleNamespace(code=self.code)], 2000) + request = self.run.orders.place.call_args.args[1] + self.assertTrue(request.order_id.startswith('zt-base-')) + deal = self.fill('base', 'b1', 100) + deal.remark = request.order_id + '|zt' + self.store.sync_state([]) + boot.sync_account_state(self.store, [], [deal]) + self.assertEqual(self.store.state[self.code]['base_qty'], 100) + if __name__ == '__main__': unittest.main()