diff --git a/py-client/libs/order.py b/py-client/libs/order.py index cead696..e488b5b 100644 --- a/py-client/libs/order.py +++ b/py-client/libs/order.py @@ -33,7 +33,7 @@ class OrderBook: """线程安全的活动委托缓存。""" def __init__( - self, order_prefix: str, lock_timeout_sec: int = 180, cancel_timeout_sec: float = 10 + 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) diff --git a/py-client/libs/orderbook.py b/py-client/libs/orderbook.py deleted file mode 100644 index e45c766..0000000 --- a/py-client/libs/orderbook.py +++ /dev/null @@ -1,159 +0,0 @@ -"""SQLite positions and deals, aligned with SDK models; one writer per database.""" - -import math -import sqlite3 -from contextlib import closing -from dataclasses import asdict, fields -from datetime import datetime -from pathlib import Path -from itertools import chain - -from sdk import DealItem, PositionItem - -SCHEMA = """ -CREATE TABLE IF NOT EXISTS positions ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - stock_code TEXT NOT NULL, - stock_name TEXT NOT NULL DEFAULT '', - direction INTEGER, - volume INTEGER NOT NULL DEFAULT 0 CHECK (volume >= 0), - open_price REAL NOT NULL DEFAULT 0, - open_cost REAL NOT NULL DEFAULT 0, - float_profit REAL NOT NULL DEFAULT 0, - market_value REAL NOT NULL DEFAULT 0, - stock_holder TEXT NOT NULL DEFAULT '', - frozen_volume INTEGER NOT NULL DEFAULT 0, - can_use_volume INTEGER NOT NULL DEFAULT 0, - on_road_volume INTEGER NOT NULL DEFAULT 0, - yesterday_volume INTEGER NOT NULL DEFAULT 0, - last_price REAL NOT NULL DEFAULT 0, - profit_rate REAL NOT NULL DEFAULT 0, - future_trade_type INTEGER, - expire_date TEXT NOT NULL DEFAULT '' -); -CREATE UNIQUE INDEX IF NOT EXISTS idx_positions_stock_code ON positions (stock_code); - -CREATE TABLE IF NOT EXISTS deals ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - stock_code TEXT NOT NULL, - order_sys_id TEXT NOT NULL CHECK (order_sys_id <> ''), - order_local_id TEXT NOT NULL CHECK (order_local_id <> ''), - 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)), - price REAL NOT NULL CHECK (price >= 0), - volume INTEGER NOT NULL CHECK (volume > 0), - trade_amount REAL NOT NULL CHECK (trade_amount > 0), - trade_date TEXT NOT NULL, - trade_time TEXT NOT NULL, - remark TEXT NOT NULL DEFAULT '', - close_profit REAL NOT NULL DEFAULT 0 -); -CREATE UNIQUE INDEX IF NOT EXISTS idx_deals_order_sys_id ON deals (order_sys_id); -CREATE INDEX IF NOT EXISTS idx_deals_order_ref ON deals (order_local_id); -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); -""" - -# SQL is assembled once; order_local_id is derived from the SDK remark property. -POSITION_FIELDS = tuple(field.name for field in fields(PositionItem)) -DEAL_FIELDS = tuple(field.name for field in fields(DealItem)) + ('order_local_id',) -POSITION_DEFAULTS = asdict(PositionItem()) -POSITION_UPSERT = ( - f"INSERT INTO positions ({', '.join(POSITION_FIELDS)}) " - f"VALUES ({', '.join(':' + key for key in POSITION_FIELDS)}) " - "ON CONFLICT(stock_code) DO UPDATE SET " - + ', '.join(f'{key} = excluded.{key}' for key in POSITION_FIELDS if key != 'stock_code') -) -DEAL_INSERT = ( - f"INSERT INTO deals ({', '.join(DEAL_FIELDS)}) " - f"VALUES ({', '.join(':' + key for key in DEAL_FIELDS)})" -) - - -class OrderBook: - """Position snapshots and append-only deals. No schema migration.""" - - def __init__(self, path: str | Path) -> None: - self.path = Path(path) - self.positions: dict[str, dict] = {} - self.deals: dict[str, dict] = {} - self.deals_sys_ids: set[str] = set() - self.path.parent.mkdir(parents=True, exist_ok=True) - with closing(self._connect()) as db: - db.executescript(SCHEMA) - self.load() - - def _connect(self) -> sqlite3.Connection: - db = sqlite3.connect(self.path, timeout=30) - db.row_factory = sqlite3.Row - return db - - def load(self) -> None: - with closing(self._connect()) as db, db: - db.execute('BEGIN') - positions = {row['stock_code']: dict(row) for row in db.execute('SELECT * FROM positions')} - deals = {row['order_sys_id']: dict(row) for row in db.execute('SELECT * FROM deals ORDER BY id')} - self.positions = positions - self.deals = deals - self.deals_sys_ids = set(deals) - - @staticmethod - def deal_record(deal: DealItem) -> dict: - if not deal.order_sys_id or deal.volume <= 0: - raise ValueError('System order ID and positive volume are required') - - row = asdict(deal) - row['order_local_id'] = deal.local_order_id - if not row['order_local_id']: - raise ValueError('Local order ID is required') - - if any(isinstance(value, float) and not math.isfinite(value) for value in row.values()): - raise ValueError('Numeric values must be finite') - amount = deal.trade_amount if deal.trade_amount > 0 else deal.price * deal.volume - if not math.isfinite(amount) or amount <= 0: - raise ValueError('Trade amount must be positive and finite') - row['trade_amount'] = amount - date = deal.trade_date or datetime.now().date().isoformat() - if len(date) == 8 and date.isdigit(): - date = f'{date[:4]}-{date[4:6]}-{date[6:]}' - row['trade_date'] = date - return row - - def sync_deals(self, deals: list[DealItem]) -> None: - new_deals = {} - for deal in deals: - if deal.order_sys_id not in self.deals_sys_ids and deal.order_sys_id not in new_deals: - new_deals[deal.order_sys_id] = self.deal_record(deal) - if not new_deals: - return - with closing(self._connect()) as db, db: - db.executemany(DEAL_INSERT, new_deals.values()) - self.load() - - def sync_positions(self, positions: list[PositionItem]) -> None: - """Replace the complete position snapshot, retaining IDs for existing stocks.""" - self.save({item.stock_code: asdict(item) for item in positions}) - - def save(self, items: dict, deals: list[dict] | None = None) -> None: - """保存持仓快照，可同时追加成交；省略 deals 时仅更新持仓。""" - new_deals = [] - if deals is not None: - if len(deals) < len(self.deals): - raise ValueError('Execution history is append-only') - new_deals = deals[len(self.deals):] - positions = [{**POSITION_DEFAULTS, **item} for item in items.values()] - for row in chain(positions, new_deals): - if any(isinstance(value, float) and not math.isfinite(value) for value in row.values()): - raise ValueError('Numeric values must be finite') - with closing(self._connect()) as db, db: - removed = [ - (row['stock_code'],) - for row in db.execute('SELECT stock_code FROM positions') - if row['stock_code'] not in items - ] - db.executemany('DELETE FROM positions WHERE stock_code = ?', removed) - db.executemany(POSITION_UPSERT, positions) - db.executemany(DEAL_INSERT, new_deals) - self.load() diff --git a/py-client/libs/state.py b/py-client/libs/state.py new file mode 100644 index 0000000..136b5cf --- /dev/null +++ b/py-client/libs/state.py @@ -0,0 +1,259 @@ +"""SQLite 策略状态与成交存储；每个数据库仅使用一个写入者，不做数据迁移。""" + +import math +import json +import logging as log +import sqlite3 +from contextlib import closing +from dataclasses import asdict, dataclass +from datetime import datetime +from pathlib import Path + +from sdk import DealItem, PositionItem + +SCHEMA = """ +-- 策略状态：base_ 表示底仓，added_ 表示补仓。 +CREATE TABLE IF NOT EXISTS state ( + id INTEGER PRIMARY KEY AUTOINCREMENT, -- 状态记录主键 + stock_code TEXT NOT NULL, -- 证券代码 + status TEXT NOT NULL DEFAULT '', -- 策略状态，由策略定义取值 + base_order_local_id TEXT NOT NULL DEFAULT '', -- 底仓本地委托编号 + base_qty INTEGER NOT NULL DEFAULT 0 CHECK (base_qty >= 0), -- 底仓数量 + base_price REAL NOT NULL DEFAULT 0, -- 底仓价格 + base_created_at TEXT NOT NULL DEFAULT '', -- 底仓创建时间 + added_order_local_id TEXT NOT NULL DEFAULT '', -- 补仓本地委托编号 + added_qty INTEGER NOT NULL DEFAULT 0 CHECK (added_qty >= 0), -- 补仓数量 + added_price REAL NOT NULL DEFAULT 0, -- 补仓价格 + added_created_at TEXT NOT NULL DEFAULT '' -- 补仓创建时间 +); +-- 每个证券仅保留一条策略状态。 +CREATE UNIQUE INDEX IF NOT EXISTS idx_state_stock_code ON state (stock_code); + +-- 首次归档前的持仓基准，清仓后仍保留，供迟到成交按时间重算。 +CREATE TABLE IF NOT EXISTS state_origin ( + stock_code TEXT PRIMARY KEY, -- 证券代码 + snapshot TEXT NOT NULL -- 初始持仓字段的 JSON 快照 +); + +-- 成交记录独立保存，不随状态删除。 +CREATE TABLE IF NOT EXISTS deals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + stock_code TEXT NOT NULL, + order_sys_id TEXT NOT NULL CHECK (order_sys_id <> ''), + order_local_id TEXT NOT NULL CHECK (order_local_id <> ''), + 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)), + price REAL NOT NULL CHECK (price >= 0), + volume INTEGER NOT NULL CHECK (volume > 0), + trade_amount REAL NOT NULL CHECK (trade_amount > 0), + trade_date TEXT NOT NULL, + trade_time TEXT NOT NULL, + remark TEXT NOT NULL DEFAULT '', + close_profit REAL NOT NULL DEFAULT 0, + is_arch INTEGER DEFAULT 0 +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_deals_order_sys_id ON deals (order_sys_id); +CREATE INDEX IF NOT EXISTS idx_deals_order_ref ON deals (order_local_id); +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); +""" + +@dataclass(slots=True) +class StateItem: + """策略状态字段；同步账户底仓时无法获知的委托编号留空。""" + + stock_code: str = '' # 证券代码 + status: str = '' # 策略状态 + base_order_local_id: str = '' # 底仓本地委托编号 + base_qty: int = 0 # 底仓数量 + base_price: float = 0.0 # 底仓价格 + base_created_at: str = '' # 底仓创建时间 + added_order_local_id: str = '' # 补仓本地委托编号 + added_qty: int = 0 # 补仓数量 + added_price: float = 0.0 # 补仓价格 + added_created_at: str = '' # 补仓创建时间 + + +class State: + """保存策略状态和只追加的成交记录，仅创建新表，不迁移旧数据。""" + + def __init__(self, path: str | Path) -> None: + self.path = Path(path) + self.state: dict[str, dict] = {} + self.deals: dict[str, dict] = {} + self.deals_sys_ids: set[str] = set() + self.path.parent.mkdir(parents=True, exist_ok=True) + with closing(self._connect()) as db: + db.executescript(SCHEMA) + self.load() + + def _connect(self) -> sqlite3.Connection: + db = sqlite3.connect(self.path, timeout=30) + db.row_factory = sqlite3.Row + return db + + def load(self) -> None: + """从数据库刷新状态、成交及去重缓存。""" + with closing(self._connect()) as db, db: + db.execute('BEGIN') + state = {row['stock_code']: dict(row) for row in db.execute('SELECT * FROM state')} + deals = {row['order_sys_id']: dict(row) for row in db.execute('SELECT * FROM deals ORDER BY id')} + self.state = state + self.deals = deals + self.deals_sys_ids = set(deals) + + def sync_deals(self, deals: list[DealItem]) -> None: + """按系统成交编号去重，整批写入成功后刷新缓存。""" + new_deals = {} + for deal in deals: + if deal.order_sys_id not in self.deals_sys_ids and deal.order_sys_id not in new_deals: + new_deals[deal.order_sys_id] = deal + if not new_deals: + return + with closing(self._connect()) as db, db: + for deal in new_deals.values(): + order_id = deal.get_local_order_id + if not order_id: + raise ValueError('Local order ID is required') + amount = deal.trade_amount if deal.trade_amount > 0 else deal.price * deal.volume + if not math.isfinite(amount) or amount <= 0: + raise ValueError('Trade amount must be positive and finite') + date = deal.trade_date or datetime.now().date().isoformat() + if len(date) == 8 and date.isdigit(): + date = f'{date[:4]}-{date[4:6]}-{date[6:]}' + # 直接读取模型字段，金额和日期的补全不修改传入模型。 + db.execute( + '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) ' + 'VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', + (deal.stock_code, deal.order_sys_id, order_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), + ) + self.load() + + def archiving(self) -> dict[str, str]: + """按证券从持仓基准重放成交；失败证券保留未归档记录并返回原因。""" + errors = {} + with closing(self._connect()) as db, db: + db.execute('BEGIN IMMEDIATE') + codes = db.execute( + 'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0 AND offset_flag IN (48, 49)' + ).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() + origin = db.execute('SELECT snapshot FROM state_origin WHERE stock_code = ?', (code,)).fetchone() + if origin is None: + # 没有旧基准时不能用已归档后的持仓反推历史，不做数据迁移。 + if db.execute( + 'SELECT 1 FROM deals WHERE stock_code = ? AND is_arch = 1 LIMIT 1', (code,) + ).fetchone(): + raise ValueError('Missing holding baseline for archived history') + state = dict(current) if current else asdict(StateItem(stock_code=code)) + db.execute('INSERT INTO state_origin VALUES (?, ?)', (code, json.dumps(state))) + else: + state = json.loads(origin['snapshot']) + # 数量相等的初始买入视为已包含在快照中，只匹配一次。 + snapshot_qty = state['base_qty'] + state['added_qty'] + covered = False + deals = db.execute( + 'SELECT * FROM deals WHERE stock_code = ? AND offset_flag IN (48, 49) ' + "ORDER BY trade_date, REPLACE(trade_time, ':', ''), id", (code,) + ).fetchall() + for deal in deals: + qty = deal['volume'] + if deal['offset_flag'] == 48: + if not covered and snapshot_qty == qty: + covered = True + continue + covered = True + total = state['added_qty'] + qty + state['added_price'] = ( + state['added_qty'] * state['added_price'] + deal['trade_amount'] + ) / total + state['added_qty'] = total + state['added_order_local_id'] = deal['order_local_id'] + state['added_created_at'] = f"{deal['trade_date']} {deal['trade_time']}".strip() + else: + # 卖出之后的买入属于新交易，不再与初始快照匹配。 + covered = True + 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: + # 重算数量和成本，保留调用方当前设置的 status 及已有记录主键。 + 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)}) " + 'ON CONFLICT(stock_code) DO UPDATE SET ' + + ', '.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 (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: + db.execute('RELEASE archive_stock') + self.load() + return errors + + def sync_state(self, positions: list[PositionItem]) -> None: + """同步完整持仓：无状态则插入底仓，已有则保留，清仓则删除。 + + 数量为零或未出现在完整持仓列表中的证券视为已清仓；空列表清空状态。 + """ + # 传入完整账户持仓；同步时间作为新增底仓的创建时间。 + created_at = datetime.now().isoformat(timespec='seconds') + holdings = {item.stock_code: item for item in positions if item.volume > 0} + with closing(self._connect()) as db, db: + db.execute('BEGIN') + existing = {row['stock_code']: dict(row) for row in db.execute('SELECT * FROM state')} + # 先保留基准，再删除清仓状态，卖出成交仍可据此归档。 + for code, row in existing.items(): + db.execute( + 'INSERT OR IGNORE INTO state_origin SELECT ?, ? WHERE NOT EXISTS ' + '(SELECT 1 FROM deals WHERE stock_code = ? AND is_arch = 1)', + (code, json.dumps(row), code), + ) + db.executemany( + 'DELETE FROM state WHERE stock_code = ?', + [(code,) for code in existing if code not in holdings], + ) + for code, item in holdings.items(): + if code in existing: + continue + if not math.isfinite(item.open_price): + raise ValueError('Base price must be finite') + # 只插入底仓字段，补仓字段使用数据库默认值。 + db.execute( + 'INSERT INTO state ' + '(stock_code, status, base_order_local_id, base_qty, base_price, base_created_at) ' + "VALUES (?, '', '', ?, ?, ?)", + (code, item.volume, item.open_price, created_at), + ) + self.load() diff --git a/py-client/main.py b/py-client/main.py index 7c85ad5..82ff8f5 100644 --- a/py-client/main.py +++ b/py-client/main.py @@ -36,7 +36,6 @@ from sdk import APIError, Client from libs.market import refresh_market from libs.collector import submit_trend_data from strategy.trend.boot import StartTrend -from strategy.zt.boot import StartZT from strategy.ipo import AutoBuyIpo @dataclass(slots=True) @@ -47,7 +46,6 @@ class StrategyDefinition: STRATEGIES = { "trend": StrategyDefinition("Trend", StartTrend), - "zt": StrategyDefinition("ZT", StartZT), } def require_windows() -> bool: diff --git a/py-client/strategy/trend/positions.py b/py-client/strategy/trend/positions.py index 78019aa..aa57c0b 100644 --- a/py-client/strategy/trend/positions.py +++ b/py-client/strategy/trend/positions.py @@ -88,16 +88,17 @@ def manage_positions( strTag = "↑" elif pnl_rate< LOSS_TIERS[0]: strTag = "↓" - - log.info( - "[Position %s ] %s %s，盈亏=%.2f%%，止盈=%s，补仓=%s", - strTag, - code, - position.stock_name, - pnl_rate, - profit_action, - loss_add_action, - ) + + if strTag != "-": + log.info( + "[Position %s ] %s %s，盈亏=%.2f%%，止盈=%s，补仓=%s", + strTag, + code, + position.stock_name, + pnl_rate, + profit_action, + loss_add_action, + ) except Exception: log.exception( "[Position] 持仓处理异常，代码=%s，继续处理后续持仓", @@ -121,15 +122,15 @@ def handle_profit( if observation.state == GridState.ARMED: return TradeDecision( False, - f"首次达到 {pnl_rate:.2f}%，峰值网格={observation.current_grid}", + f"首次, PNL:{pnl_rate:.2f}%，网格={observation.current_grid}", ) if observation.state == GridState.RAISED: return TradeDecision( False, - f"上涨至 {pnl_rate:.2f}%，峰值网格={observation.current_grid}", + f"突破, PNL:{pnl_rate:.2f}%，网格={observation.current_grid}", ) if observation.state == GridState.STEADY: - return TradeDecision(False) + return TradeDecision(False,f"持平, PNL:{pnl_rate:.2f}%，网格={observation.current_grid}",) if runtime.orders.busy(position.stock_code, "SELL"): return TradeDecision(False, "卖出委托处理中") diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index 2880c1a..1c0b782 100644 --- a/py-client/strategy/zt/boot.py +++ b/py-client/strategy/zt/boot.py @@ -20,7 +20,6 @@ from libs.overview import Overview from libs.order import BUSY_STATUSES, OrderBook from libs.watch import DipWatch from libs.runtime import Runtime -from .state import TState, SOLD from .open import open_signal from .positions import manage_positions @@ -32,10 +31,10 @@ def StartZT() -> None: config.global_config.qmt_token, config.HTTP_TIMEOUT, ) as client: - state = TState( - Path(config.global_config.qmt_data_dir) - / f"zt_{config.account_config.account_id}_state.db" - ) + # state = TState( + # Path(config.global_config.qmt_data_dir) + # / f"zt_{config.account_config.account_id}_state.db" + # ) executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="zt") run = Runtime( client=client, diff --git a/py-client/strategy/zt/open.py b/py-client/strategy/zt/open.py index ee53224..46584dc 100644 --- a/py-client/strategy/zt/open.py +++ b/py-client/strategy/zt/open.py @@ -8,10 +8,9 @@ from libs.calc import calc_buy_volume from sdk import OP_BUY from libs.runtime import Runtime from libs.order import PlaceOrderRequest -from .state import TState -def open_signal(run: Runtime, state: TState, ticks, signals, available: float) -> float: +def open_signal(run: Runtime, ticks, signals, available: float) -> float: """逐个验证开仓信号并提交买入委托，返回本轮剩余资金。""" for item in signals: try: @@ -20,9 +19,6 @@ def open_signal(run: Runtime, state: TState, ticks, signals, available: float) - break if item.code in run.account_cfg.excluded_codes: continue - item_state = state.items.get(item.code) - if item_state is not None and item_state.base_qty > 0: - continue # 由委托簿检查活动委托，防止重复下单。 if ( run.orders.busy(item.code, "BUY") diff --git a/py-client/strategy/zt/positions.py b/py-client/strategy/zt/positions.py index 19d3f41..72dd200 100644 --- a/py-client/strategy/zt/positions.py +++ b/py-client/strategy/zt/positions.py @@ -7,12 +7,10 @@ from libs.grid_take_profit import GridState from sdk import OP_BUY, OP_SELL, PositionItem from libs.order import PlaceOrderRequest from libs.runtime import Runtime -from .state import READY, SOLD, TState def manage_positions( run: Runtime, - state_store: TState, ticks, positions: list[PositionItem], available: float, diff --git a/py-client/strategy/zt/state.py b/py-client/strategy/zt/state.py deleted file mode 100644 index cb16246..0000000 --- a/py-client/strategy/zt/state.py +++ /dev/null @@ -1,179 +0,0 @@ -"""做 T 策略的持仓状态和逐笔实际成交记录。""" - -import math -from dataclasses import dataclass -from datetime import datetime -from pathlib import Path - -from sdk import DealItem, PositionItem -from libs.orderbook import OrderBook - -READY, SOLD, DONE = "READY", "SOLD", "DONE" - - -@dataclass(slots=True) -class TStateItem: - code: str - base_qty: int = 0 - base_cost: float = 0.0 - trade_date: str = "" - phase: str = READY - sell_qty: int = 0 - sell_price: float = 0.0 - buy_qty: int = 0 - buy_cost: float = 0.0 - id: int = 0 - base_order_id: str = '' - added_order_id: str = '' - added_num: int = 0 - added_qty: int = 0 - added_cost: float = 0.0 - - -class TState: - """Apply actual executions immediately, atomically with their position changes.""" - - def __init__(self, path: str | Path) -> None: - self._store = OrderBook(path) - self.path = self._store.path - self._load() - - @staticmethod - def _is_zt_deal(deal: DealItem) -> bool: - return ( - deal.side == 'BUY' and deal.local_order_id.startswith(('zt-base-', 'zt-t-buy-')) - ) or ( - deal.side == 'SELL' and deal.local_order_id.startswith('zt-t-sell-') - ) - - @staticmethod - def _reset(item: TStateItem, date: str) -> bool: - if item.phase == DONE and item.trade_date != date: - item.phase, item.trade_date = READY, '' - item.sell_qty = item.buy_qty = 0 - item.sell_price = item.buy_cost = 0.0 - return True - return False - - @classmethod - def _apply_t_deal(cls, item: TStateItem, deal: dict) -> None: - """实时入账与重启恢复共用同一套做 T 轮次计算。""" - cls._reset(item, deal['trade_date']) - qty, amount = deal['volume'], deal['trade_amount'] - if str(deal['offset_flag']) in ('24', '49'): - total = item.sell_qty + qty - item.sell_price = (item.sell_qty * item.sell_price + amount) / total - item.sell_qty = total - item.phase = SOLD - else: - total = item.buy_qty + qty - item.buy_cost = (item.buy_qty * item.buy_cost + amount) / total - item.buy_qty = total - item.phase = DONE if total >= item.sell_qty else SOLD - item.trade_date = deal['trade_date'] - - def reconcile( - self, positions: list[PositionItem], deals: list[DealItem] - ) -> None: - """Deduplicate each fill; partial fills do not wait for order completion.""" - today = datetime.now().date().isoformat() - seen = {row['order_sys_id'] for row in self.deals} - rows = [] - for deal in deals: - if not self._is_zt_deal(deal) or deal.order_sys_id in seen: - continue - try: - row = self._store.deal_record(deal) - except ValueError: - continue - rows.append(row) - seen.add(deal.order_sys_id) - rows.sort(key=lambda r: (r['trade_date'], r['trade_time'])) - modified = False - try: - # Snapshot includes these fills: subtract their net quantity before replay. - net = {} - for row in rows: - net[row['stock_code']] = net.get(row['stock_code'], 0) + ( - row['volume'] if str(row['offset_flag']) in ('23', '48') else -row['volume'] - ) - for position in positions: - code = position.stock_code - if code in self.items or position.volume <= 0: - continue - if not math.isfinite(position.open_price) or position.open_price <= 0: - continue - qty = max(0, position.volume - net.get(code, 0)) - self.items[code] = TStateItem(code, qty, position.open_price if qty else 0.0) - modified = True - - # 全部卖出时快照可能已无该证券，按净卖出数量恢复待买回的底仓数量。 - for code, delta in net.items(): - if code not in self.items and delta < 0: - self.items[code] = TStateItem(code, -delta) - - for row in rows: - code = row['stock_code'] - item = self.items.get(code) - if item is None: - item = self.items[code] = TStateItem(code) - self._reset(item, row['trade_date']) - qty, amount = row['volume'], row['trade_amount'] - if row['order_local_id'].startswith('zt-base-'): - total = item.base_qty + qty - item.base_cost = (item.base_qty * item.base_cost + amount) / total - item.base_qty = total - item.base_order_id = row['order_local_id'] - else: - self._apply_t_deal(item, row) - self.deals.append(row) - modified = True - - for item in self.items.values(): - modified = self._reset(item, today) or modified - if modified: - self.save() - except Exception: - self._load() - raise - - def save(self) -> None: - try: - self._store.save( - { - code: { - 'stock_code': item.code, - 'volume': item.base_qty, - 'open_price': item.base_cost, - 'open_cost': item.base_qty * item.base_cost, - } - for code, item in self.items.items() - }, - self.deals, - ) - except Exception: - self._load() - raise - - def _load(self) -> None: - self._store.load() - self.items = {} - for code, position in self._store.positions.items(): - self.items[code] = TStateItem(code, position['volume'], position['open_price'], id=position['id']) - self.deals = [ - {key: value for key, value in deal.items() if key != 'id'} - for deal in self._store.deals.values() - ] - # 轮次明细不占用持仓表字段，从已保存的逐笔成交重建。 - for deal in self.deals: - item = self.items.get(deal['stock_code']) - if item is None: - continue - local_order_id = deal['order_local_id'] - if local_order_id.startswith('zt-base-'): - item.base_order_id = local_order_id - else: - self._apply_t_deal(item, deal) - today = datetime.now().date().isoformat() - for item in self.items.values(): - self._reset(item, today) diff --git a/py-client/tests/test_deal_model.py b/py-client/tests/test_deal_model.py index 34d465b..be5e690 100644 --- a/py-client/tests/test_deal_model.py +++ b/py-client/tests/test_deal_model.py @@ -7,7 +7,7 @@ from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock from libs.order import OrderBook as ActiveOrders -from libs.orderbook import OrderBook +from libs.state import State from sdk.models import Assets, DealItem, OrderItem, PositionItem from sdk.portfolio import PortfolioMixin @@ -76,16 +76,21 @@ class ApiModelTests(unittest.TestCase): deal = self.client.deals()[0] with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / 'state.db' - book = OrderBook(path) + book = State(path) book.sync_deals([deal]) - loaded = OrderBook(path).deals['sys1'] + loaded = State(path).deals['sys1'] self.assertEqual(loaded['volume'], deal.volume) self.assertEqual(loaded['trade_date'], '2026-09-07') - deal.trade_amount = 0 - self.assertEqual(OrderBook.deal_record(deal)['trade_amount'], 1000) - deal.price = 0 - with self.assertRaises(ValueError): - OrderBook.deal_record(deal) + deal.order_sys_id = 'sys2' + deal.trade_amount = 0 + book.sync_deals([deal, deal]) + self.assertEqual(book.deals['sys2']['trade_amount'], 1000) + self.assertEqual(deal.trade_amount, 0) + deal.order_sys_id = 'sys3' + deal.price = 0 + with self.assertRaises(ValueError): + book.sync_deals([deal]) + self.assertEqual(set(State(path).deals), {'sys1', 'sys2'}) if __name__ == '__main__': diff --git a/py-client/tests/test_orderbook.py b/py-client/tests/test_orderbook.py index af1b039..f90e3cc 100644 --- a/py-client/tests/test_orderbook.py +++ b/py-client/tests/test_orderbook.py @@ -7,7 +7,7 @@ from pathlib import Path from datetime import datetime from unittest.mock import patch -from libs.orderbook import OrderBook +from libs.state import State, StateItem from sdk import DealItem, PositionItem from strategy.zt.state import DONE, READY, SOLD, TState @@ -63,14 +63,14 @@ class OrderBookTests(unittest.TestCase): def test_json_is_never_read(self): legacy = self.path.with_suffix('.json') legacy.write_text('invalid JSON', encoding='utf-8') - book = OrderBook(self.path) + book = State(self.path) self.assertIsNone(book.load()) - self.assertEqual((book.positions, book.deals, book.deals_sys_ids), ({}, {}, set())) + self.assertEqual((book.items, book.deals, book.deals_sys_ids), ({}, {}, set())) self.assertEqual(legacy.read_text(encoding='utf-8'), 'invalid JSON') def test_sync_deals_deduplicates_batch_and_restart(self): - book = OrderBook(self.path) - self.assertEqual((book.positions, book.deals, book.deals_sys_ids), ({}, {}, set())) + book = State(self.path) + self.assertEqual((book.items, book.deals, book.deals_sys_ids), ({}, {}, set())) first = self.deal('base', 'd1', 40, 10, '20260901') second = self.deal('base', 'd2', 60, 12) book.sync_deals([first, first, second]) @@ -78,7 +78,7 @@ class OrderBookTests(unittest.TestCase): self.assertEqual(book.deals['d1']['trade_date'], '2026-09-01') self.assertEqual(book.deals['d2']['volume'], 60) self.assertEqual(book.deals['d2']['order_local_id'], 'zt-base-order1') - book = OrderBook(self.path) + book = State(self.path) self.assertEqual(book.deals_sys_ids, {'d1', 'd2'}) self.assertEqual(book.deals['d1']['order_local_id'], 'zt-base-order1') with patch.object(book, '_connect') as connect: @@ -88,7 +88,7 @@ class OrderBookTests(unittest.TestCase): self.assertEqual(len(book.deals), 2) def test_sync_deals_failure_rolls_back_entire_batch_and_cache(self): - book = OrderBook(self.path) + book = State(self.path) first = self.deal('base', 'd1', 100, 10) invalid = self.deal('base', 'd2', 100, 10) invalid.offset_flag = -1 @@ -96,18 +96,18 @@ class OrderBookTests(unittest.TestCase): book.sync_deals([first, invalid]) self.assertEqual(book.deals, {}) self.assertEqual(book.deals_sys_ids, set()) - self.assertEqual(OrderBook(self.path).deals, {}) + self.assertEqual(State(self.path).deals, {}) invalid.offset_flag = 23 book.sync_deals([first, invalid]) self.assertEqual(book.deals_sys_ids, {'d1', 'd2'}) def test_load_refreshes_all_caches(self): - book = OrderBook(self.path) - writer = OrderBook(self.path) - writer.sync_positions([PositionItem(stock_code='600000.SH', volume=100)]) + book = State(self.path) + writer = State(self.path) + writer.sync_state([PositionItem(stock_code='600000.SH', volume=100)]) writer.sync_deals([self.deal('base', 'd1', 100, 10)]) book.load() - self.assertEqual(book.positions['600000.SH']['volume'], 100) + self.assertEqual(book.items['600000.SH']['base_qty'], 100) self.assertEqual(book.deals_sys_ids, {'d1'}) self.assertEqual(book.deals['d1']['remark'], 'zt-base-order1|zt') @@ -140,11 +140,11 @@ class OrderBookTests(unittest.TestCase): state.reconcile([], [self.deal('base', 'd1', 100, 10)]) with closing(sqlite3.connect(self.path)) as db: tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} - self.assertEqual(tables, {'positions', 'deals', 'sqlite_sequence'}) + self.assertEqual(tables, {'state', 'deals', 'sqlite_sequence'}) columns = {row[1] for row in db.execute('PRAGMA table_info(deals)')} - self.assertEqual(columns, {'id', 'order_local_id', *(field.name for field in fields(DealItem))}) + self.assertEqual(columns, {'id', 'order_local_id', 'is_arch', *(field.name for field in fields(DealItem))}) indexes = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='index'")} - self.assertTrue({'idx_positions_stock_code', 'idx_deals_order_sys_id', + self.assertTrue({'idx_state_stock_code', 'idx_deals_order_sys_id', 'idx_deals_order_ref', 'idx_deals_stock_code_date', 'idx_deals_date_time'} <= indexes) for index, expected in ( ('idx_deals_order_sys_id', ['order_sys_id']), @@ -165,29 +165,62 @@ class OrderBookTests(unittest.TestCase): self.assertEqual(state.items['600000.SH'].base_cost, 10) def test_position_columns_defaults_indexes_and_stable_id(self): - store = OrderBook(self.path) + store = State(self.path) store.sync_deals([self.deal('base', 'd1', 100, 10)]) saved_deals = dict(store.deals) with closing(sqlite3.connect(self.path)) as db: - columns = {row[1] for row in db.execute('PRAGMA table_info(positions)')} - self.assertEqual(columns, {'id', *(field.name for field in fields(PositionItem))}) - indexes = {row[1] for row in db.execute('PRAGMA index_list(positions)')} - self.assertEqual(indexes, {'idx_positions_stock_code'}) + columns = {row[1] for row in db.execute('PRAGMA table_info(state)')} + self.assertEqual(columns, {'id', *(field.name for field in fields(StateItem))}) + indexes = {row[1] for row in db.execute('PRAGMA index_list(state)')} + self.assertEqual(indexes, {'idx_state_stock_code'}) position = PositionItem(stock_code='600000.SH', volume=100, open_price=10, stock_name='stock', can_use_volume=100, float_profit=-2.5) - store.sync_positions([position]) - saved = store.positions[position.stock_code] + store.sync_state([position]) + saved = store.items[position.stock_code] first_id = saved['id'] - self.assertEqual({k: v for k, v in saved.items() if k != 'id'}, asdict(position)) + self.assertEqual(saved['base_qty'], 100) + self.assertEqual(saved['base_price'], 10) + self.assertEqual(saved['added_qty'], 0) + self.assertEqual(saved['base_order_local_id'], '') + self.assertTrue(saved['base_created_at']) position.volume = 200 - store.sync_positions([position]) - self.assertEqual(store.positions[position.stock_code]['id'], first_id) - self.assertEqual(store.positions[position.stock_code]['volume'], 200) - store.sync_positions([]) - self.assertEqual(store.positions, {}) + position.open_price = 12 + store.sync_state([position]) + self.assertEqual(store.items[position.stock_code]['id'], first_id) + self.assertEqual(store.items[position.stock_code], saved) + self.assertEqual(State(self.path).items[position.stock_code], saved) + store.sync_state([position, PositionItem(stock_code='600001.SH', volume=100)]) + self.assertEqual(store.items[position.stock_code], saved) + self.assertEqual(store.items['600001.SH']['base_qty'], 100) + position.volume = 0 + store.sync_state([position, PositionItem(stock_code='600002.SH')]) + self.assertEqual(store.items, {}) + self.assertEqual(State(self.path).items, {}) + store.sync_state([PositionItem(stock_code='600001.SH', volume=100)]) + self.assertGreater(store.items['600001.SH']['id'], first_id) + store.sync_state([]) + self.assertEqual(store.items, {}) self.assertEqual(store.deals, saved_deals) - store.sync_positions([PositionItem(stock_code='600001.SH')]) - self.assertGreater(store.positions['600001.SH']['id'], first_id) + + def test_state_fields_survive_restart_and_sync(self): + book = State(self.path) + row = asdict(StateItem( + stock_code='600000.SH', status='READY', + base_order_local_id='base-1', base_qty=100, base_price=10, + base_created_at='2026-09-08T09:30:00', + added_order_local_id='added-1', added_qty=50, added_price=9, + added_created_at='2026-09-08T10:30:00', + )) + book.save({row['stock_code']: row}) + saved = book.items[row['stock_code']] + self.assertEqual({k: v for k, v in saved.items() if k != 'id'}, row) + book = State(self.path) + book.sync_state([PositionItem(stock_code=row['stock_code'], volume=150, open_price=9.5)]) + self.assertEqual(book.items[row['stock_code']], saved) + row['added_qty'] = -1 + with self.assertRaises(sqlite3.IntegrityError): + book.save({row['stock_code']: row}) + self.assertEqual(State(self.path).items[row['stock_code']], saved) def test_base_split_fills_and_snapshot_do_not_double_count(self): state = TState(self.path) diff --git a/py-client/tests/test_state_archiving.py b/py-client/tests/test_state_archiving.py new file mode 100644 index 0000000..65a7a38 --- /dev/null +++ b/py-client/tests/test_state_archiving.py @@ -0,0 +1,183 @@ +import sqlite3 +import tempfile +import unittest +from contextlib import closing +from pathlib import Path + +from libs.state import State +from sdk import DealItem, PositionItem + + +class ArchivingTests(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.book = State(Path(tmp.name) / 'state.db') + + def insert_deal(self, order, qty, amount, time, code='600000.SH', flag=48): + with closing(self.book._connect()) as db, db: + db.execute( + 'INSERT INTO deals (stock_code, order_sys_id, order_local_id, offset_flag, ' + 'price, volume, trade_amount, trade_date, trade_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', + (code, order, order, flag, amount / qty, qty, amount, '2026-09-08', time), + ) + + def test_accumulates_once_and_preserves_base(self): + self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) + original = dict(self.book.state['600000.SH']) + self.insert_deal('first', 40, 400, '10:00:00') + self.insert_deal('second', 60, 720, '10:01:00') + self.book.archiving() + row = self.book.state['600000.SH'] + self.assertEqual(row['added_qty'], 100) + self.assertAlmostEqual(row['added_price'], 11.2) + self.assertEqual(row['added_order_local_id'], 'second') + self.assertEqual(row['added_created_at'], '2026-09-08 10:01:00') + for key in original: + if not key.startswith('added_'): + self.assertEqual(row[key], original[key]) + self.assertTrue(all(deal['is_arch'] == 1 for deal in self.book.deals.values())) + restarted = State(self.book.path) + restarted.archiving() + self.assertEqual(restarted.state, self.book.state) + self.insert_deal('late', 50, 500, '09:59:00') + self.book.archiving() + row = self.book.state['600000.SH'] + self.assertEqual(row['added_qty'], 150) + self.assertAlmostEqual(row['added_price'], 1620 / 150) + self.assertEqual(row['added_order_local_id'], 'second') + + 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') + self.insert_deal('buy', 60, 600, '10:00:00') + self.insert_deal('partial', 40, 480, '10:01:00', flag=49) + self.book.archiving() + row = self.book.state['600000.SH'] + self.assertEqual((row['base_qty'], row['base_price'], row['added_qty'], row['added_price']), + (100, 8, 60, 10)) + self.insert_deal('sell_added', 60, 720, '10:02:00', flag=49) + self.book.archiving() + row = self.book.state['600000.SH'] + self.assertEqual((row['base_qty'], row['added_qty'], row['added_price']), (100, 0, 0)) + self.assertEqual((row['added_order_local_id'], row['added_created_at']), ('', '')) + self.insert_deal('sell_base', 100, 1200, '10:03:00', flag=49) + self.book.archiving() + self.assertEqual(self.book.state, {}) + self.book.archiving() + self.assertEqual(State(self.book.path).state, {}) + self.assertTrue(all(deal['is_arch'] == 1 for deal in self.book.deals.values())) + + def test_sell_crosses_into_base_then_liquidates(self): + self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) + self.insert_deal('buy', 50, 500, '10:00:00') + self.insert_deal('sell', 80, 960, '10:01:00', flag=49) + self.book.archiving() + row = self.book.state['600000.SH'] + self.assertEqual((row['base_qty'], row['base_price'], row['added_qty']), (70, 8, 0)) + self.insert_deal('buy_again', 30, 300, '10:02:00') + self.insert_deal('sell_all', 100, 1200, '10:03:00', flag=49) + self.book.archiving() + self.assertEqual(self.book.state, {}) + + def test_excess_sell_rolls_back_and_other_directions_are_skipped(self): + self.insert_deal('other', 100, 1000, '09:59:00', flag=23) + self.book.archiving() + self.assertEqual(self.book.state, {}) + self.assertEqual(self.book.deals['other']['is_arch'], 0) + 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) + restarted = State(self.book.path) + self.assertEqual(restarted.state, {}) + self.assertTrue(all(deal['is_arch'] == 0 for deal in restarted.deals.values())) + + def test_new_state_and_failed_mark_roll_back_together(self): + self.insert_deal('first', 100, 1000, '10:00:00') + self.insert_deal('second', 100, 1200, '10:01:00', code='600001.SH') + self.book.load() + with closing(self.book._connect()) as db: + 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) + restarted = State(self.book.path) + self.assertEqual(set(restarted.state), {'600000.SH'}) + self.assertEqual(restarted.state, self.book.state) + self.assertEqual(restarted.deals['first']['is_arch'], 1) + self.assertEqual(restarted.deals['second']['is_arch'], 0) + with closing(self.book._connect()) as db: + db.execute('DROP TRIGGER fail_archive') + self.book.archiving() + self.assertEqual(len(self.book.state), 2) + self.assertEqual(self.book.state['600000.SH']['base_qty'], 0) + self.assertEqual(self.book.state['600000.SH']['added_qty'], 100) + + def test_snapshot_matching_buy_is_not_added(self): + self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) + self.book.sync_deals([DealItem( + stock_code='600000.SH', order_sys_id='first', remark='base1|test', + offset_flag=48, volume=100, price=8, trade_amount=800, + trade_date='20260908', trade_time='100000', + )]) + self.assertEqual(self.book.archiving(), {}) + row = self.book.state['600000.SH'] + self.assertEqual((row['base_qty'], row['added_qty']), (100, 0)) + self.assertEqual(self.book.deals['first']['is_arch'], 1) + restarted = State(self.book.path) + self.assertEqual(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(), {}) + row = self.book.state['600000.SH'] + self.assertEqual((row['base_qty'], row['added_qty'], row['status']), (100, 100, 'CUSTOM')) + + def test_old_archived_history_without_baseline_is_not_reapplied(self): + self.insert_deal('old', 100, 1000, '10:00:00') + 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.assertIn('baseline', errors['600000.SH']) + self.assertEqual(self.book.deals['old']['is_arch'], 1) + self.assertEqual(self.book.deals['new']['is_arch'], 0) + + def test_late_buy_replays_after_liquidation_and_restart(self): + self.insert_deal('buy', 100, 1000, '10:00:00') + self.insert_deal('sell', 100, 1500, '10:02:00', flag=49) + self.book.archiving() + self.assertEqual(self.book.state, {}) + self.book = State(self.book.path) + self.insert_deal('late', 100, 2000, '100100') + self.assertEqual(self.book.archiving(), {}) + row = self.book.state['600000.SH'] + self.assertEqual(row['added_qty'], 100) + self.assertEqual(row['added_price'], 15) + + def test_snapshot_deletion_before_sell_archiving(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.book.sync_state([]) + self.book = State(self.book.path) + self.assertEqual(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()) + 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.assertEqual(self.book.deals['bad_sell']['is_arch'], 1) + self.assertNotIn('600000.SH', self.book.state) + + +if __name__ == '__main__': + unittest.main()