fix zt&state.py

This commit is contained in:
2026-09-12 13:42:23 +08:00
parent bcb03e2ed9
commit 7a7049ce44
9 changed files with 677 additions and 363 deletions

View File

@@ -4,12 +4,16 @@ import math
import logging as log
import sqlite3
from contextlib import closing
from dataclasses import asdict, dataclass
from dataclasses import asdict, dataclass, fields
from datetime import datetime
from pathlib import Path
from sdk import DealItem, PositionItem
FLAG_BUY = 48
FLAG_SELL = 49
UNATTRIBUTED_PREFIX = '__unattributed__:'
SCHEMA = """
-- 策略状态base_ 表示底仓added_ 表示补仓。
CREATE TABLE IF NOT EXISTS state (
@@ -69,14 +73,29 @@ class StateItem:
added_created_at: str = '' # 补仓创建时间
_STATE_COLUMNS = tuple(field.name for field in fields(StateItem))
_UPSERT_STATE = (
f"INSERT INTO state ({', '.join(_STATE_COLUMNS)}) "
f"VALUES ({', '.join(':' + key for key in _STATE_COLUMNS)}) "
"ON CONFLICT(stock_code) DO UPDATE SET "
+ ', '.join(f'{key} = excluded.{key}' for key in _STATE_COLUMNS if key != 'stock_code')
)
class State:
"""保存策略状态和只追加的成交记录,仅创建新表,不迁移旧数据。"""
"""单写入者使用的 SQLite 存储;公开缓存仅在事务成功后替换。
sync_state 同步持仓基准sync_deals 保存成交archiving 记入增量。
同证券未归档成交全部为买入且数量等于当前总持仓时,视为已计入
快照,只标记归档;其他成交作为增量处理。本类不做表结构迁移。
"""
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.blocked_codes: set[str] = set()
self.path.parent.mkdir(parents=True, exist_ok=True)
with closing(self._connect()) as db:
db.executescript(SCHEMA)
@@ -87,176 +106,242 @@ class State:
db.row_factory = sqlite3.Row
return db
def get_by_code(self,code: str) -> dict:
s = self.state.get(code,{})
return s
@staticmethod
def _read_state(db: sqlite3.Connection) -> dict[str, dict]:
return {row['stock_code']: dict(row) for row in db.execute('SELECT * FROM state')}
@staticmethod
def _read_deals(db: sqlite3.Connection) -> dict[str, dict]:
return {
row['order_sys_id']: dict(row)
for row in db.execute('SELECT * FROM deals ORDER BY id')
}
def get_by_code(self, code: str) -> dict:
"""返回缓存中的状态;不存在时返回空字典。"""
return self.state.get(code, {})
def load(self) -> None:
"""缓存状态表和成交记录"""
self.load_state()
self.load_deals()
def load_state(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')}
state = self._read_state(db)
deals = self._read_deals(db)
self.state, self.deals, self.deals_sys_ids = state, deals, set(deals)
def load_state(self) -> None:
"""只刷新状态缓存。"""
with closing(self._connect()) as db, db:
db.execute('BEGIN')
state = self._read_state(db)
self.state = state
def load_deals(self) -> None:
"""缓存成交记录"""
"""只刷新成交缓存及其去重编号集合"""
with closing(self._connect()) as db, db:
db.execute('BEGIN')
deals = {row['order_sys_id']: dict(row) for row in db.execute('SELECT * FROM deals ORDER BY id')}
self.deals = deals
self.deals_sys_ids = set(deals)
deals = self._read_deals(db)
self.deals, self.deals_sys_ids = deals, set(deals)
def sync_deals(self, deals: list[DealItem]) -> None:
"""按成交编号去重;初始化底仓时,已包含在快照内的成交可直接标记归档。"""
new_deals = {}
@staticmethod
def _insert_deals(db: sqlite3.Connection, deals: list[DealItem]) -> None:
"""共享事务内保存成交;空本地编号使用明确的待核对标记。"""
existing = {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 self.deals_sys_ids and deal.order_sys_id not in new_deals:
new_deals[deal.order_sys_id] = deal
if deal.order_sys_id not in existing:
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:
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, is_arch) '
'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, 0),
)
self.load_deals()
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)
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')
db.execute('BEGIN IMMEDIATE')
existing = {row['stock_code'] for row in db.execute('SELECT stock_code FROM state')}
db.executemany(
'DELETE FROM state WHERE stock_code = ?',
[(code,) for code in existing if code not in holdings],
)
values = []
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_state()
values.append((code, item.volume, item.open_price, created_at))
db.executemany(
'DELETE FROM state WHERE stock_code = ?',
[(code,) for code in existing if code not in holdings],
)
db.executemany(
'INSERT INTO state (stock_code, base_qty, base_price, base_created_at) '
'VALUES (?, ?, ?, ?)', values,
)
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 _archive_pending(self, db: sqlite3.Connection, snapshot_dedup: bool = True) -> None:
"""保留现有逐证券保存点ZT 增量路径禁用数量相等推断。"""
codes = db.execute(
'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0'
).fetchall()
for row in codes:
code = row['stock_code']
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')
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'
).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,)
).fetchall()
for deal in deals:
qty = deal['volume']
if deal['offset_flag'] in (23, 48):
# 按已有的 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)}) "
'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 (23, 24, 48, 49)', (code,)
)
except (ValueError, sqlite3.IntegrityError) as exc:
# 撤销这只股票的全部归档修改,成交仍保持未归档,供下次重试。
db.execute('ROLLBACK TO archive_stock')
log.warning('[归档] %s 失败,保留未归档成交:%s', code, exc)
finally:
# 释放当前股票的回滚点;整个事务在退出外层 with 时提交。
db.execute('RELEASE archive_stock')
# 数据库提交完成后刷新内存缓存,让策略读到最新持仓和归档标记。
self.load()
self._archive_pending(db)
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')
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_deals(db, deals)
self._archive_pending(db, snapshot_dedup=False)
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}
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)))