diff --git a/api/qmt_rest_new.py b/api/qmt_rest_new.py index c952772..cd2888d 100644 --- a/api/qmt_rest_new.py +++ b/api/qmt_rest_new.py @@ -174,7 +174,7 @@ class PortfolioHandler(BaseHandler): result = { "assets": format_assets(account_data), "positions": format_holding(positions), - "orders": [fixed_fields(order) for order in orders], + "orders": format_orders(orders), } self.write_json(result) @@ -183,8 +183,7 @@ class PortfolioHandler(BaseHandler): class HoldingHandler(BaseHandler): def get(self): positions = safe_call(get_trade_detail_data, self.acc(), 'stock', 'position') or [] - holding = format_holding(positions) - self.write_json({"data": holding}) + self.write_json(format_holding(positions)) class OrgHandler(BaseHandler): def get(self, handler_type): @@ -207,22 +206,18 @@ class OrgHandler(BaseHandler): # get_trade_detail_data('account') - Query account assets class AssetsHandler(BaseHandler): def get(self): - _data = safe_call(get_trade_detail_data, self.acc(), 'stock', 'account') + _data = safe_call(get_trade_detail_data, self.acc(), 'stock', 'account') or [] self.write_json(format_assets(_data)) class OrderHandler(BaseHandler): def get(self): - ret = safe_call(get_trade_detail_data, self.acc(), 'stock', 'order') - if ret is None: - ret = [] - result = [fixed_fields(obj) for obj in ret] - self.write_json(result) + ret = safe_call(get_trade_detail_data, self.acc(), 'stock', 'order') or [] + self.write_json(format_orders(ret)) class DealHandler(BaseHandler): def get(self): deals = safe_call(get_trade_detail_data, self.acc(), 'stock', 'deal') or [] - rets = [fixed_fields(deal) for deal in deals] - self.write_json({"deals": rets}) + self.write_json(format_deals(deals)) # ContextInfo.get_full_tick() - Get full tick data class FullTickHandler(BaseHandler): @@ -330,56 +325,70 @@ def format_holding(positions): for position in positions: stock = position.m_strInstrumentID + '.' + position.m_strExchangeID holding[stock] = { - 'StockCode': stock, - 'TradeID':position.m_strTradeID, - 'StockName': position.m_strInstrumentName, - 'Direction': position.m_nDirection, - 'Volume': position.m_nVolume, - 'OpenPrice': position.m_dOpenPrice, - 'OpenCost':position.m_dOpenCost, - 'FloatProfit': position.m_dFloatProfit, - 'MarketValue': position.m_dMarketValue, - 'StockHolder': position.m_strStockHolder, - 'FrozenVolume': position.m_nFrozenVolume, - 'CanUseVolume': position.m_nCanUseVolume, - 'OnRoadVolume': position.m_nOnRoadVolume, - 'YesterdayVolume': position.m_nYesterdayVolume, - 'LastPrice': position.m_dLastPrice, - 'ProfitRate': position.m_dProfitRate, - 'FutureTradeType': position.m_eFutureTradeType, - 'ExpireDate': position.m_strExpireDate + 'stock_code': stock, + 'stock_name': position.m_strInstrumentName, + 'direction': position.m_nDirection, + 'volume': position.m_nVolume, + 'open_price': position.m_dOpenPrice, + 'open_cost':position.m_dOpenCost, + 'float_profit': position.m_dFloatProfit, + 'market_value': position.m_dMarketValue, + 'stock_holder': position.m_strStockHolder, + 'frozen_volume': position.m_nFrozenVolume, + 'can_use_volume': position.m_nCanUseVolume, + 'on_road_volume': position.m_nOnRoadVolume, + 'yesterday_volume': position.m_nYesterdayVolume, + 'last_price': position.m_dLastPrice, + 'profit_rate': position.m_dProfitRate, + 'future_trade_type': position.m_eFutureTradeType, + 'expire_date': position.m_strExpireDate } return holding -TRADE_DETAIL_FIELDS = ( - 'm_strOrderSysID', 'm_strInstrumentID', 'm_strExchangeID', - 'm_strInstrumentName', 'm_nOffsetFlag', 'm_nOrderStatus', - 'm_nVolumeTotal', 'm_nVolumeTraded', 'm_nOrderTime', - 'm_strInsertDate', 'm_strInsertTime', 'm_strRemark', - 'm_dPrice', 'm_dTradePrice', 'm_dTradeAmount', -) -MISSING = object() - - -def fixed_fields(obj, fields=TRADE_DETAIL_FIELDS): - result = {} - for field in fields: - try: - value = getattr(obj, field, MISSING) - except TypeError: - continue - if value is MISSING: - continue - if not callable(value): - result[field] = str(value) - if not result: - attrs = getattr(obj, '__dict__', {}) - result = { - key: str(value) for key, value in attrs.items() - if not key.startswith('_') and not callable(value) - } +def format_orders(orders): + """显式转换委托/成交字段，保留原始数值类型。""" + result = [] + for order in orders: + result.append({ + 'stock_code': order.m_strInstrumentID + '.' + order.m_strExchangeID, + 'order_sys_id': order.m_strOrderSysID, + 'ref': order.m_nRef, + 'order_ref': order.m_strOrderRef, + 'direction': order.m_nDirection, + 'offset_flag': order.m_nOffsetFlag, + 'limit_price': order.m_dLimitPrice, + 'volume_total_original': order.m_nVolumeTotalOriginal, + 'volume_traded': order.m_nVolumeTraded, + 'volume_total': order.m_nVolumeTotal, + 'traded_price': order.m_dTradedPrice, + 'trade_amount': order.m_dTradeAmount, + 'insert_date': order.m_strInsertDate, + 'insert_time': order.m_strInsertTime, + 'remark': order.m_strRemark, + 'order_status': order.m_nOrderStatus, + }) return result +def format_deals(deals): + """显式转换委托/成交字段，保留原始数值类型。""" + result = [] + for d in deals: + result.append({ + 'stock_code': d.m_strInstrumentID + '.' + d.m_strExchangeID, + 'order_sys_id': d.m_strOrderSysID, + 'ref': d.m_nRef, + 'order_ref': d.m_strOrderRef, + 'direction': d.m_nDirection, + 'offset_flag': d.m_nOffsetFlag, + 'price': d.m_dPrice, + 'volume': d.m_nVolume, + 'trade_amount': d.m_dTradeAmount, + 'trade_date': d.m_strTradeDate, + 'trade_time': d.m_strTradeTime, + 'remark': d.m_strRemark, + 'close_profit': d.m_dCloseProfit, + }) + return result # ============= Route registration ============= def make_app(): diff --git a/docs/zt.md b/docs/zt.md index d331b01..5bdc136 100644 --- a/docs/zt.md +++ b/docs/zt.md @@ -19,39 +19,22 @@ zt_max_price: 200 ## SQLite 状态存储 `libs/orderbook.py` 使用标准库 `sqlite3`，数据库路径为 -`{qmt_data_dir}/zt_{account_id}_state.db`，每个账户/策略独立存储，由一个策略实例串行更新。 -启动时仅创建当前表结构和索引，不执行数据迁移或旧 JSON 导入。 +`{qmt_data_dir}/zt_{account_id}_state.db`，每个账户/策略由单个实例串行更新。 +启动时仅创建当前表结构和索引，不执行迁移或旧 JSON 导入。 -`OrderBook` 初始化后调用 `load()`，填充 `positions`（按代码）、`deals`（按系统订单号） -和 `deals_sys_ids`（系统订单号集合）。`sync_deals(list[DealItem])` 根据该集合过滤已保存及 -同批重复成交，再以单个事务批量插入，提交成功后刷新缓存；不修改持仓。 -`sys_order_id` 对应 API 的 `m_strOrderSysID`，`local_order_id` 从 `m_strRemark` 的首段提取。 +两表使用 SDK 同名字段；另有自增主键 `id`，成交表增加从 `remark` 提取的 `order_local_id`： -| 表 | 字段与用途 | 索引 | +| 表 | 数据模型 | 索引 | | --- | --- | --- | -| `positions` | 自增 `id`、股票代码 `code`、底仓订单/数量/成本 `base_order_id/base_qty/base_cost`、补仓订单/次数/数量/成本 `added_order_id/added_num/added_qty/added_cost`、状态 `status` | `code` 唯一索引；`base_order_id`；`added_order_id` | -| `deals` | `id`、系统/本地订单号、证券信息、方向、API 状态、剩余/成交数量、委托日期时间、备注、委托价格及成交均价/金额，字段映射见下表 | `id` 主键；`sys_order_id` 唯一；`local_order_id`；`(code, insert_date)`；`(insert_date, insert_time)` | +| `positions` | `PositionItem`：`stock_code`、`stock_name`、`direction`、`volume`、`open_price`、`open_cost`、`float_profit`、`market_value`、`stock_holder`、`frozen_volume`、`can_use_volume`、`on_road_volume`、`yesterday_volume`、`last_price`、`profit_rate`、`future_trade_type`、`expire_date` | `stock_code` 唯一索引 | +| `deals` | `DealItem`：`stock_code`、`order_sys_id`、`ref`、`order_ref`、`direction`、`offset_flag`、`price`、`volume`、`trade_amount`、`trade_date`、`trade_time`、`remark`、`close_profit`，以及 `order_local_id` | `order_sys_id` 唯一索引；`order_local_id`；`stock_code`；`trade_date` | -SDK 的 `DealItem` 与成交表业务字段一致： +`load()` 只更新 `positions`、`deals`、`deals_sys_ids` 缓存，无返回值。 +`sync_positions(list[PositionItem])` 保存完整持仓快照，同一证券更新时保留自增 ID。 +`sync_deals(list[DealItem])` 按系统订单号去重后批量写入；同批重复记录仅写一次。 +`order_local_id` 取 `DealItem.local_order_id`（`remark` 首段），为空时拒绝写入。 +成交日期规范为 `YYYY-MM-DD`，金额缺失时用成交价格乘数量补足。 -| API 字段 | SDK / SQLite 字段 | -| --- | --- | -| `m_strOrderSysID` | `sys_order_id` | -| `m_strInstrumentID`、`m_strExchangeID` | `instrument_id`、`exchange_id`，组合生成 `code` | -| `m_strInstrumentName` | `name` | -| `m_nOffsetFlag` | `offset_flag`，解析生成买卖方向 `side` | -| `m_nOrderStatus` | `status` | -| `m_nVolumeTotal`、`m_nVolumeTraded` | `remaining_volume`、`traded_volume` | -| `m_nOrderTime` | `order_time` | -| `m_strInsertDate`、`m_strInsertTime` | `insert_date`、`insert_time` | -| `m_strRemark` | `remark`，提取 `local_order_id` | -| `m_dPrice`、`m_dTradePrice`、`m_dTradeAmount` | `price`、`trade_price`、`trade_amount` | - -API 返回已成交数据，每个系统订单号仅入库一次；`status` 保存 API 原值，不增加确认流程。 -成交金额缺失时仅使用成交均价乘成交数量补足，不使用委托价格。 -买卖方向使用 `side`，ZT 通过本地委托号前缀区分底仓买入与做 T 买回。 -ZT 根据实际成交数量及金额更新持仓，日期取 API 提供的 `insert_date`，入库时规范为 `YYYY-MM-DD`。 -首次接管持仓仅写持仓表。活动委托和重复下单检查由 `libs/order.py` 的委托簿负责。 -持仓快照与新增成交在同一事务内保存，历史成交只追加；写入失败回滚数据库并恢复内存状态。 -持仓更新按 `code` 保留原有自增 ID。ZT 将轮次状态写入 `status`，卖出、买回数量及均价、 -轮次日期在重启时从 `deals` 重建；补仓字段预留，ZT 买回不计为补仓。 +ZT 使用 `volume/open_price` 保存底仓数量与成本，做 T 轮次从成交历史恢复， +不再使用持仓表的旧状态、底仓订单或补仓字段。买卖方向由 `offset_flag` 计算， +本地订单号从 `remark` 提取。持仓与新增成交在同一事务提交，失败时回滚并恢复内存。 diff --git a/py-client/libs/__pycache__/collector.cpython-311.pyc b/py-client/libs/__pycache__/collector.cpython-311.pyc index bb23612..ea480ca 100644 Binary files a/py-client/libs/__pycache__/collector.cpython-311.pyc and b/py-client/libs/__pycache__/collector.cpython-311.pyc differ diff --git a/py-client/libs/__pycache__/order.cpython-311.pyc b/py-client/libs/__pycache__/order.cpython-311.pyc new file mode 100644 index 0000000..d7b3905 Binary files /dev/null and b/py-client/libs/__pycache__/order.cpython-311.pyc differ diff --git a/py-client/libs/__pycache__/orderbook.cpython-311.pyc b/py-client/libs/__pycache__/orderbook.cpython-311.pyc new file mode 100644 index 0000000..8ee035d Binary files /dev/null and b/py-client/libs/__pycache__/orderbook.cpython-311.pyc differ diff --git a/py-client/libs/__pycache__/overview.cpython-311.pyc b/py-client/libs/__pycache__/overview.cpython-311.pyc new file mode 100644 index 0000000..8ae6cf7 Binary files /dev/null and b/py-client/libs/__pycache__/overview.cpython-311.pyc differ diff --git a/py-client/libs/__pycache__/runtime.cpython-311.pyc b/py-client/libs/__pycache__/runtime.cpython-311.pyc new file mode 100644 index 0000000..5664a47 Binary files /dev/null and b/py-client/libs/__pycache__/runtime.cpython-311.pyc differ diff --git a/py-client/libs/__pycache__/watch.cpython-311.pyc b/py-client/libs/__pycache__/watch.cpython-311.pyc new file mode 100644 index 0000000..e7a4396 Binary files /dev/null and b/py-client/libs/__pycache__/watch.cpython-311.pyc differ diff --git a/py-client/libs/collector.py b/py-client/libs/collector.py index 56f3fae..16ac94e 100644 --- a/py-client/libs/collector.py +++ b/py-client/libs/collector.py @@ -9,6 +9,15 @@ import httpx COLLECTOR_URL = "http://139.224.247.176:13499/collector" +def submit_trend_data() -> None: + """每五分钟提交趋势策略的最新缓存，尚无快照时跳过。""" + from strategy.trend.boot import get_collector_snapshot + + snapshot = get_collector_snapshot() + if snapshot is not None: + collector_push(*snapshot) + + def _json_value(value: Any) -> Any: """Convert the QMT model values into values accepted by a JSON encoder.""" if is_dataclass(value) and not isinstance(value, type): @@ -27,7 +36,7 @@ def _json_value(value: Any) -> Any: def collector_push(account_id: str, assets: Any, positions: Any) -> None: - """Best-effort collector upload; never propagate errors to the caller.""" + """[暂停] 数据收集提交，太耗时，超过200毫秒.""" try: payload = _json_value( { diff --git a/py-client/libs/order.py b/py-client/libs/order.py index 8a66d41..9e725bf 100644 --- a/py-client/libs/order.py +++ b/py-client/libs/order.py @@ -70,29 +70,29 @@ class OrderBook: for item in orders: # 不处理状态不对的 - if item.status not in TRACKED_STATUSES: + if str(item.order_status) not in TRACKED_STATUSES: continue - if item.status in BUSY_STATUSES: - busy_keys.add(self._busy_key(item.side, item.code)) + if str(item.order_status) in BUSY_STATUSES: + busy_keys.add(self._busy_key(item.side, item.stock_code)) # 清理过期的 if ( item.created_at is not None and item.local_order_id.startswith(f"{self.order_prefix}-") - and item.status in CANCELABLE_STATUSES + and str(item.order_status) in CANCELABLE_STATUSES and current - item.created_at > self.cancel_timeout_sec ): try: - client.cancel_by_id(item.id) + client.cancel_by_id(item.order_sys_id) canceled += 1 logging.info( "[Order] 超时撤单，代码=%s，方向=%s，柜台订单=%s", - item.code, + item.stock_code, item.side, - item.id, + item.order_sys_id, ) except Exception: logging.exception( - "[Order] 撤单失败，保留在途状态，订单=%s", item.id + "[Order] 撤单失败，保留在途状态，订单=%s", item.order_sys_id ) # 缓存本次有效订单 diff --git a/py-client/libs/orderbook.py b/py-client/libs/orderbook.py index 8a4219c..5adef1b 100644 --- a/py-client/libs/orderbook.py +++ b/py-client/libs/orderbook.py @@ -1,62 +1,79 @@ -"""SQLite 状态簿：每个账户/策略使用独立数据库，单个策略串行读写。""" +"""SQLite positions and deals, aligned with SDK models; one writer per database.""" from __future__ import annotations import math import sqlite3 from contextlib import closing -from dataclasses import asdict +from dataclasses import asdict, fields from datetime import datetime from pathlib import Path -from sdk import DealItem +from sdk import DealItem, PositionItem SCHEMA = """ CREATE TABLE IF NOT EXISTS positions ( id INTEGER PRIMARY KEY AUTOINCREMENT, - code TEXT NOT NULL, - base_order_id TEXT NOT NULL DEFAULT '', - base_qty INTEGER NOT NULL DEFAULT 0 CHECK (base_qty >= 0), - base_cost REAL NOT NULL DEFAULT 0.0 CHECK (base_cost >= 0), - added_order_id TEXT NOT NULL DEFAULT '', - added_num INTEGER NOT NULL DEFAULT 0 CHECK (added_num >= 0), - added_qty INTEGER NOT NULL DEFAULT 0 CHECK (added_qty >= 0), - added_cost REAL NOT NULL DEFAULT 0.0 CHECK (added_cost >= 0), - status TEXT NOT NULL + 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_code ON positions (code); -CREATE INDEX IF NOT EXISTS idx_positions_base_order_id ON positions (base_order_id); -CREATE INDEX IF NOT EXISTS idx_positions_added_order_id ON positions (added_order_id); +CREATE UNIQUE INDEX IF NOT EXISTS idx_positions_stock_code ON positions (stock_code); CREATE TABLE IF NOT EXISTS deals ( - id INTEGER PRIMARY KEY, - sys_order_id TEXT NOT NULL UNIQUE CHECK (sys_order_id <> ''), - local_order_id TEXT NOT NULL, - code TEXT NOT NULL, - instrument_id TEXT NOT NULL, - exchange_id TEXT NOT NULL, - name TEXT NOT NULL, - offset_flag TEXT NOT NULL, - side TEXT NOT NULL CHECK (side IN ('BUY', 'SELL')), - status TEXT NOT NULL, - remaining_volume INTEGER NOT NULL CHECK (remaining_volume >= 0), - traded_volume INTEGER NOT NULL CHECK (traded_volume > 0), - order_time INTEGER NOT NULL, - insert_date TEXT NOT NULL, - insert_time TEXT NOT NULL, - remark TEXT NOT NULL, - price REAL NOT NULL, - trade_price REAL NOT NULL CHECK (trade_price >= 0), - trade_amount REAL NOT NULL CHECK (trade_amount > 0) + 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 INDEX IF NOT EXISTS idx_deals_local_order_id ON deals (local_order_id); -CREATE INDEX IF NOT EXISTS idx_deals_code_date ON deals (code, insert_date); -CREATE INDEX IF NOT EXISTS idx_deals_date_time ON deals (insert_date, insert_time); +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_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: - """Persist position snapshots and append-only executions; one writer per database.""" + """Position snapshots and append-only deals. No schema migration.""" def __init__(self, path: str | Path) -> None: self.path = Path(path) @@ -76,8 +93,8 @@ class OrderBook: def load(self) -> None: with closing(self._connect()) as db, db: db.execute('BEGIN') - positions = {row['code']: dict(row) for row in db.execute('SELECT * FROM positions')} - deals = {row['sys_order_id']: dict(row) for row in db.execute('SELECT * FROM deals ORDER BY id')} + 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) @@ -85,86 +102,58 @@ class OrderBook: @staticmethod def _insert_deals(db: sqlite3.Connection, deals: list[dict]) -> None: - db.executemany( - """INSERT INTO deals - (sys_order_id, local_order_id, code, instrument_id, exchange_id, name, - offset_flag, side, status, remaining_volume, traded_volume, order_time, - insert_date, insert_time, remark, price, trade_price, trade_amount) - VALUES (:sys_order_id, :local_order_id, :code, :instrument_id, :exchange_id, :name, - :offset_flag, :side, :status, :remaining_volume, :traded_volume, :order_time, - :insert_date, :insert_time, :remark, :price, :trade_price, :trade_amount)""", - deals, - ) + db.executemany(DEAL_INSERT, deals) @staticmethod def deal_record(deal: DealItem) -> dict: - """API 字段转入库记录；委托价格不用于推算成交金额。""" - if not deal.sys_order_id or deal.traded_volume <= 0: - raise ValueError('System order ID and positive traded volume are required') + if not deal.order_sys_id or deal.volume <= 0: + raise ValueError('System order ID and positive volume are required') + if not deal.local_order_id: + raise ValueError('Local order ID is required') row = asdict(deal) + row['order_local_id'] = deal.local_order_id if any(isinstance(value, float) and not math.isfinite(value) for value in row.values()): - raise ValueError('Prices and amounts must be finite') - amount = deal.trade_amount if deal.trade_amount > 0 else deal.trade_price * deal.traded_volume + 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('Execution amount must be positive and finite') + raise ValueError('Trade amount must be positive and finite') row['trade_amount'] = amount - date = deal.insert_date or datetime.now().date().isoformat() + 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['insert_date'] = date + row['trade_date'] = date return row def sync_deals(self, deals: list[DealItem]) -> None: - """按系统订单号去重，批量写入已成交数据；失败时不更新缓存。""" new_deals = [] seen = self.deals_sys_ids.copy() for deal in deals: - if deal.sys_order_id in seen: + if deal.order_sys_id in seen: continue new_deals.append(self.deal_record(deal)) - seen.add(deal.sys_order_id) + seen.add(deal.order_sys_id) if not new_deals: return with closing(self._connect()) as db, db: self._insert_deals(db, new_deals) 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}, list(self.deals.values())) + def save(self, items: dict, deals: list[dict]) -> None: if len(deals) < self._deal_count: raise ValueError('Execution history is append-only') new_deals = deals[self._deal_count:] - positions = [ - { - 'base_order_id': '', 'base_qty': 0, 'base_cost': 0.0, - 'added_order_id': '', 'added_num': 0, 'added_qty': 0, 'added_cost': 0.0, - **item, - } - for item in items.values() - ] - for row in [*items.values(), *new_deals]: + positions = [{**asdict(PositionItem()), **item} for item in items.values()] + for row in [*positions, *new_deals]: if any(isinstance(value, float) and not math.isfinite(value) for value in row.values()): - raise ValueError('Quantities, prices and amounts must be finite') + raise ValueError('Numeric values must be finite') with closing(self._connect()) as db, db: - # 更新已有证券时保留其自增 ID；仅删除快照中已移除的证券。 - for row in db.execute('SELECT code FROM positions').fetchall(): - if row['code'] not in items: - db.execute('DELETE FROM positions WHERE code = ?', (row['code'],)) - db.executemany( - """INSERT INTO positions - (code, base_order_id, base_qty, base_cost, - added_order_id, added_num, added_qty, added_cost, status) - VALUES (:code, :base_order_id, :base_qty, :base_cost, - :added_order_id, :added_num, :added_qty, :added_cost, :status) - ON CONFLICT(code) DO UPDATE SET - base_order_id = excluded.base_order_id, - base_qty = excluded.base_qty, - base_cost = excluded.base_cost, - added_order_id = excluded.added_order_id, - added_num = excluded.added_num, - added_qty = excluded.added_qty, - added_cost = excluded.added_cost, - status = excluded.status""", - positions, - ) + for row in db.execute('SELECT stock_code FROM positions').fetchall(): + if row['stock_code'] not in items: + db.execute('DELETE FROM positions WHERE stock_code = ?', (row['stock_code'],)) + db.executemany(POSITION_UPSERT, positions) self._insert_deals(db, new_deals) self.load() diff --git a/py-client/libs/overview.py b/py-client/libs/overview.py index e56b7b2..4036197 100644 --- a/py-client/libs/overview.py +++ b/py-client/libs/overview.py @@ -26,8 +26,7 @@ def Overview(assets, positions, account_cfg=None) -> None: if position.volume <= 0: continue log.info( - "[启动] %s %s %s，持仓=%d，可用=%d，成本=%.2f(%.2f)，现价=%.2f，盈亏=%.2f%%", - position.trade_id, + "[启动] %s %s，持仓=%d，可用=%d，成本=%.2f(%.2f)，现价=%.2f，盈亏=%.2f%%", position.stock_code, position.stock_name, position.volume, diff --git a/py-client/main.py b/py-client/main.py index 0ca4dab..7c85ad5 100644 --- a/py-client/main.py +++ b/py-client/main.py @@ -34,6 +34,7 @@ logging.getLogger("httpx").setLevel(logging.WARNING) 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 @@ -109,6 +110,7 @@ def wait_for_any_key() -> None: def main() -> int: + scheduler = None try: if not require_windows(): logging.error("本程序仅支持 Windows 环境运行") @@ -143,18 +145,29 @@ def main() -> int: replace_existing=True, next_run_time=datetime.now(), ) + scheduler.add_job( + submit_trend_data, + trigger="interval", + minutes=5, + id="trend_collector", + replace_existing=True, + ) + logging.info("趋势策略数据提交任务已注册：每5分钟读取缓存提交") scheduler.start() logging.info("IPO 自动打新定时任务已启动：每日 10:00、14:00") logging.info("大盘信号后台刷新已启动：每分钟一次") STRATEGIES[config.account_config.strategy].start_strategy() - logging.info("%s 策略启动成功",config.account_config.strateg) + logging.info("%s 策略已结束", config.account_config.strategy) return 0 except (OSError, yaml.YAMLError, ValueError, RuntimeError, KeyError) as e: print(f"启动失败: {e}", file=sys.stderr, flush=True) traceback.print_exception(type(e), e, e.__traceback__) wait_for_any_key() return 1 + finally: + if scheduler is not None and scheduler.running: + scheduler.shutdown(wait=True) if __name__ == "__main__": diff --git a/py-client/sdk/__pycache__/__init__.cpython-311.pyc b/py-client/sdk/__pycache__/__init__.cpython-311.pyc index a33517f..0d113c2 100644 Binary files a/py-client/sdk/__pycache__/__init__.cpython-311.pyc and b/py-client/sdk/__pycache__/__init__.cpython-311.pyc differ diff --git a/py-client/sdk/__pycache__/models.cpython-311.pyc b/py-client/sdk/__pycache__/models.cpython-311.pyc index 556142d..7e15380 100644 Binary files a/py-client/sdk/__pycache__/models.cpython-311.pyc and b/py-client/sdk/__pycache__/models.cpython-311.pyc differ diff --git a/py-client/sdk/__pycache__/portfolio.cpython-311.pyc b/py-client/sdk/__pycache__/portfolio.cpython-311.pyc index 665186b..a04fd18 100644 Binary files a/py-client/sdk/__pycache__/portfolio.cpython-311.pyc and b/py-client/sdk/__pycache__/portfolio.cpython-311.pyc differ diff --git a/py-client/sdk/models.py b/py-client/sdk/models.py index 822baad..3180177 100644 --- a/py-client/sdk/models.py +++ b/py-client/sdk/models.py @@ -13,118 +13,75 @@ def _number(value: Any, kind: type = float) -> Any: @dataclass(slots=True) class OrderItem: - """由 QMT 委托明细解析得到的标准订单记录。""" + """Fields match format_orders() exactly.""" - id: str - code: str - side: str - remark: str - status: str - created_at: datetime | None - volume: int - local_order_id: str = "" - traded_volume: int = 0 - remaining_volume: int = 0 - exchange_id: str = "" - name: str = "" - price: float = 0.0 - trade_price: float = 0.0 + stock_code: str = "" + order_sys_id: str = "" + ref: int = 0 + order_ref: str = "" + direction: int = 0 + offset_flag: int = 0 + limit_price: float = 0.0 + volume_total_original: int = 0 + volume_traded: int = 0 + volume_total: int = 0 + traded_price: float = 0.0 trade_amount: float = 0.0 + insert_date: str = "" + insert_time: str = "" + remark: str = "" + order_status: int = 0 - @classmethod - def from_trade_detail(cls, data: dict[str, Any]) -> "OrderItem": - """从 TradeDetailData 的 QMT 原始字段创建订单。""" - instrument_id = str(data.get("m_strInstrumentID") or "") - exchange_id = str(data.get("m_strExchangeID") or "") - code = ( - f"{instrument_id}.{exchange_id}" - if instrument_id and exchange_id - else instrument_id - ) - remaining_volume = _number(data.get("m_nVolumeTotal"), int) - traded_volume = _number(data.get("m_nVolumeTraded"), int) - remark = str(data.get("m_strRemark") or "") - return cls( - id=str(data.get("m_strOrderSysID") or ""), - code=code, - side={"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"}.get( - str(data.get("m_nOffsetFlag")), "" - ), - remark=remark, - status=str(data.get("m_nOrderStatus") or ""), - created_at=_trade_datetime(data), - volume=remaining_volume + traded_volume, - local_order_id=remark.split("|", 1)[0] if remark else "", - traded_volume=traded_volume, - remaining_volume=remaining_volume, - exchange_id=exchange_id, - name=str(data.get("m_strInstrumentName") or ""), - price=_number(data.get("m_dPrice")), - trade_price=_number(data.get("m_dTradePrice")), - trade_amount=_number(data.get("m_dTradeAmount")), - ) + @property + def side(self) -> str: + return _side(self.offset_flag) + + @property + def local_order_id(self) -> str: + return self.remark.split("|", 1)[0] + + @property + def created_at(self) -> datetime | None: + return _parse_datetime(self.insert_date, self.insert_time) @dataclass(slots=True) class DealItem: - """Execution data from the API's fixed order-detail fields.""" + """Fields match format_deals() exactly.""" - sys_order_id: str = "" - local_order_id: str = "" - code: str = "" - instrument_id: str = "" - exchange_id: str = "" - name: str = "" - offset_flag: str = "" - side: str = "" - status: str = "" - remaining_volume: int = 0 - traded_volume: int = 0 - order_time: int = 0 - insert_date: str = "" - insert_time: str = "" - remark: str = "" + stock_code: str = "" + order_sys_id: str = "" + ref: int = 0 + order_ref: str = "" + direction: int = 0 + offset_flag: int = 0 price: float = 0.0 - trade_price: float = 0.0 + volume: int = 0 trade_amount: float = 0.0 + trade_date: str = "" + trade_time: str = "" + remark: str = "" + close_profit: float = 0.0 - @classmethod - def from_trade_detail(cls, data: dict[str, Any]) -> "DealItem": - instrument_id = str(data.get("m_strInstrumentID") or "") - exchange_id = str(data.get("m_strExchangeID") or "") - remark = str(data.get("m_strRemark") or "") - offset_flag = str(data.get("m_nOffsetFlag") or "") - return cls( - sys_order_id=str(data.get("m_strOrderSysID") or ""), - local_order_id=remark.split("|", 1)[0] if remark else "", - code=f"{instrument_id}.{exchange_id}" if instrument_id and exchange_id else instrument_id, - instrument_id=instrument_id, - exchange_id=exchange_id, - name=str(data.get("m_strInstrumentName") or ""), - offset_flag=offset_flag, - side={"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"}.get(offset_flag, ""), - status=str(data.get("m_nOrderStatus") or ""), - remaining_volume=_number(data.get("m_nVolumeTotal"), int), - traded_volume=_number(data.get("m_nVolumeTraded"), int), - order_time=_number(data.get("m_nOrderTime"), int), - insert_date=str(data.get("m_strInsertDate") or ""), - insert_time=str(data.get("m_strInsertTime") or ""), - remark=remark, - price=_number(data.get("m_dPrice")), - trade_price=_number(data.get("m_dTradePrice")), - trade_amount=_number(data.get("m_dTradeAmount")), - ) + @property + def side(self) -> str: + return _side(self.offset_flag) + + @property + def local_order_id(self) -> str: + return self.remark.split("|", 1)[0] @dataclass(slots=True) class PositionItem: + """Fields match format_holding() exactly.""" + stock_code: str = "" stock_name: str = "" - trade_id: str = "" direction: Any = None volume: int = 0 - open_cost: float = 0.0 open_price: float = 0.0 + open_cost: float = 0.0 float_profit: float = 0.0 market_value: float = 0.0 stock_holder: str = "" @@ -137,67 +94,14 @@ class PositionItem: future_trade_type: Any = None expire_date: str = "" - @classmethod - def from_dict(cls, data: dict[str, Any], code: str = "") -> "PositionItem": - return cls( - stock_code=str(data.get("StockCode") or code), - stock_name=str(data.get("StockName") or ""), - trade_id=str(data.get("TradeID") or ""), - open_cost=_number(data.get("OpenCost")), - direction=data.get("Direction"), - volume=_number(data.get("Volume"), int), - open_price=_number(data.get("OpenPrice")), - float_profit=_number(data.get("FloatProfit")), - market_value=_number(data.get("MarketValue")), - stock_holder=str(data.get("StockHolder") or ""), - frozen_volume=_number(data.get("FrozenVolume"), int), - can_use_volume=_number(data.get("CanUseVolume"), int), - on_road_volume=_number(data.get("OnRoadVolume"), int), - yesterday_volume=_number(data.get("YesterdayVolume"), int), - last_price=_number(data.get("LastPrice")), - profit_rate=_number(data.get("ProfitRate")), - future_trade_type=data.get("FutureTradeType"), - expire_date=str(data.get("ExpireDate") or ""), - ) - - @classmethod - def from_trade_detail(cls, data: dict[str, Any]) -> "PositionItem": - """从 TradeDetailData/Holding 的 QMT 原始字段创建持仓。""" - return cls( - stock_code=str(data.get("StockCode") or ""), - stock_name=str(data.get("StockName") or ""), - trade_id=str(data.get("TradeID") or ""), - direction=data.get("Direction"), - volume=_number(data.get("Volume"), int), - open_cost=_number(data.get("OpenCost")), - open_price=_number(data.get("OpenPrice")), - float_profit=_number(data.get("FloatProfit")), - market_value=_number(data.get("MarketValue")), - stock_holder=str(data.get("StockHolder") or ""), - frozen_volume=_number(data.get("FrozenVolume"), int), - can_use_volume=_number(data.get("CanUseVolume"), int), - on_road_volume=_number(data.get("OnRoadVolume"), int), - yesterday_volume=_number(data.get("YesterdayVolume"), int), - last_price=_number(data.get("LastPrice")), - profit_rate=_number(data.get("ProfitRate")), - future_trade_type=data.get("FutureTradeType"), - expire_date=str(data.get("ExpireDate") or ""), - ) - @dataclass(slots=True) class Assets: + """Fields match format_assets() exactly.""" + total: float = 0.0 available: float = 0.0 - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "Assets": - """兼容新版 QMT 原始资金字段及旧版简化字段。""" - return cls( - total=_number(data.get("m_dBalance", data.get("total"))), - available=_number(data.get("m_dAvailable", data.get("available"))), - ) - @dataclass(slots=True) class Portfolio: @@ -206,17 +110,14 @@ class Portfolio: orders: list[OrderItem] -def _trade_datetime(data: dict[str, Any]) -> datetime | None: - return _parse_datetime( - str(data.get("m_strInsertDate") or ""), - str(data.get("m_strInsertTime") or ""), - ) +def _side(offset_flag: int) -> str: + return {"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"}.get(str(offset_flag), "") def _parse_datetime(date: str, clock: str) -> datetime | None: clock = clock.replace(":", "").zfill(6) try: - return datetime.strptime(date + clock, "%Y%m%d%H%M%S") + return datetime.strptime(date.replace("-", "") + clock, "%Y%m%d%H%M%S") except ValueError: return None diff --git a/py-client/sdk/portfolio.py b/py-client/sdk/portfolio.py index 53e117d..fef2b69 100644 --- a/py-client/sdk/portfolio.py +++ b/py-client/sdk/portfolio.py @@ -17,33 +17,33 @@ class PortfolioMixin: def portfolio(self) -> Portfolio: data = self._get_json("/api/portfolio") or {} positions = { - code: PositionItem.from_dict(value, code) + code: PositionItem(**value) for code, value in data.get("positions", {}).items() } return Portfolio( - assets=Assets.from_dict(data.get("assets", {})), + assets=Assets(**data.get("assets", {})), positions=positions, - orders=[OrderItem.from_trade_detail(row) for row in data.get("orders", [])], + orders=[OrderItem(**row) for row in data.get("orders", [])], ) def positions(self) -> tuple[list[str], list[PositionItem]]: data = self._get_json("/api/portfolio/positions") or {} positions = [ - PositionItem.from_dict(value, code) - for code, value in data.get("data", {}).items() + PositionItem(**value) + for value in data.values() ] - return [item.stock_code for item in positions], positions + return list(data), positions def assets(self) -> Assets: - return Assets.from_dict(self._get_json("/api/portfolio/assets") or {}) + return Assets(**(self._get_json("/api/portfolio/assets") or {})) def orders(self) -> list[OrderItem]: data = self._get_json("/api/portfolio/order") or [] - return [OrderItem.from_trade_detail(row) for row in data] + return [OrderItem(**row) for row in data] def deals(self) -> list[DealItem]: data = self._get_json("/api/portfolio/deal") or [] - return [DealItem.from_trade_detail(row) for row in data] + return [DealItem(**row) for row in data] def trade_detail_data(self, datatype: str) -> Any: datatype = str(datatype).strip().lower() diff --git a/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc b/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc index aea9cea..5b4167f 100644 Binary files a/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc and b/py-client/strategy/trend/__pycache__/boot.cpython-311.pyc differ diff --git a/py-client/strategy/trend/__pycache__/open.cpython-311.pyc b/py-client/strategy/trend/__pycache__/open.cpython-311.pyc index 2cd8ab3..3d6b1a0 100644 Binary files a/py-client/strategy/trend/__pycache__/open.cpython-311.pyc and b/py-client/strategy/trend/__pycache__/open.cpython-311.pyc differ diff --git a/py-client/strategy/trend/__pycache__/positions.cpython-311.pyc b/py-client/strategy/trend/__pycache__/positions.cpython-311.pyc index 6b08e8c..8225d93 100644 Binary files a/py-client/strategy/trend/__pycache__/positions.cpython-311.pyc and b/py-client/strategy/trend/__pycache__/positions.cpython-311.pyc differ diff --git a/py-client/strategy/trend/boot.py b/py-client/strategy/trend/boot.py index 13018e8..4da149f 100644 --- a/py-client/strategy/trend/boot.py +++ b/py-client/strategy/trend/boot.py @@ -8,15 +8,16 @@ from __future__ import annotations import time import logging as log from concurrent.futures import Future, ThreadPoolExecutor +from copy import deepcopy from datetime import datetime +from threading import Lock import config from libs.calc import trading_time from libs.market import market_allow_open from libs.overview import Overview from libs.signal import init_signals, SignalItem -from libs.collector import collector_push -from sdk import Client +from sdk import Assets, Client, PositionItem from libs.grid_take_profit import GridTrailingTracker from libs.order import OrderBook from libs.watch import DipWatch @@ -24,6 +25,23 @@ from libs.runtime import Runtime from .open import open_signal from .positions import manage_positions +_collector_lock = Lock() +_collector_snapshot: tuple[str, Assets, list[PositionItem]] | None = None + + +def _cache_portfolio(account_id: str, assets: Assets, positions: list[PositionItem]) -> None: + """整体替换最新快照，策略线程不执行序列化和网络上报。""" + global _collector_snapshot + with _collector_lock: + _collector_snapshot = (account_id, assets, positions) + + +def get_collector_snapshot() -> tuple[str, Assets, list[PositionItem]] | None: + """供 scheduler 读取；复制在锁外执行，不阻塞下一轮缓存更新。""" + with _collector_lock: + snapshot = _collector_snapshot + return deepcopy(snapshot) + def StartTrend() -> None: """初始化趋势策略，并以 30 秒间隔持续执行。""" @@ -37,6 +55,7 @@ def StartTrend() -> None: portfolio = client.portfolio() assets = portfolio.assets positions = list(portfolio.positions.values()) + _cache_portfolio(config.account_config.account_id, assets, positions) order_book = OrderBook("trend") order_book.refresh(client, portfolio.orders) @@ -51,7 +70,7 @@ def StartTrend() -> None: len(signals), len(positions), ) - executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="trend") + executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="trend") run = Runtime( client=client, global_cfg=config.global_config, @@ -116,22 +135,13 @@ def RunOnce(run: Runtime, signals: list[SignalItem]) -> None: assets = portfolio.assets position_codes = list(portfolio.positions) positions = list(portfolio.positions.values()) + _cache_portfolio(run.account_cfg.account_id, assets, positions) run.orders.refresh(run.client, portfolio.orders) except Exception: log.exception("[Portfolio] 刷新账户快照失败") return - futures: list[tuple[str, Future]] = [ - ( - "数据提交", - run.executor.submit( - collector_push, - run.account_cfg.account_id, - assets, - positions, - ), - ) - ] + futures: list[tuple[str, Future]] = [] # 2. 验证可用资金；低于资金安全线时禁止开新仓。 allow_open_by_cash = ( diff --git a/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc b/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc index 4810a0e..1518a6b 100644 Binary files a/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc and b/py-client/strategy/zt/__pycache__/boot.cpython-311.pyc differ diff --git a/py-client/strategy/zt/__pycache__/open.cpython-311.pyc b/py-client/strategy/zt/__pycache__/open.cpython-311.pyc index 01a2f16..015b5e0 100644 Binary files a/py-client/strategy/zt/__pycache__/open.cpython-311.pyc and b/py-client/strategy/zt/__pycache__/open.cpython-311.pyc differ diff --git a/py-client/strategy/zt/__pycache__/positions.cpython-311.pyc b/py-client/strategy/zt/__pycache__/positions.cpython-311.pyc index 941d0df..409aaf0 100644 Binary files a/py-client/strategy/zt/__pycache__/positions.cpython-311.pyc and b/py-client/strategy/zt/__pycache__/positions.cpython-311.pyc differ diff --git a/py-client/strategy/zt/__pycache__/state.cpython-311.pyc b/py-client/strategy/zt/__pycache__/state.cpython-311.pyc index 2636d00..996f1a3 100644 Binary files a/py-client/strategy/zt/__pycache__/state.cpython-311.pyc and b/py-client/strategy/zt/__pycache__/state.cpython-311.pyc differ diff --git a/py-client/strategy/zt/state.py b/py-client/strategy/zt/state.py index fbf6cf8..b21f0bc 100644 --- a/py-client/strategy/zt/state.py +++ b/py-client/strategy/zt/state.py @@ -60,9 +60,9 @@ class TState: @classmethod def _apply_t_deal(cls, item: TStateItem, deal: dict) -> None: """实时入账与重启恢复共用同一套做 T 轮次计算。""" - cls._reset(item, deal['insert_date']) - qty, amount = deal['traded_volume'], deal['trade_amount'] - if deal['side'] == 'SELL': + 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 @@ -72,32 +72,32 @@ class TState: 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['insert_date'] + 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['sys_order_id'] for row in self.deals} + seen = {row['order_sys_id'] for row in self.deals} rows = [] for deal in deals: - if not self._is_zt_deal(deal) or deal.sys_order_id in seen: + 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.sys_order_id) - rows.sort(key=lambda r: (r['insert_date'], r['insert_time'])) + 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['code']] = net.get(row['code'], 0) + ( - row['traded_volume'] if row['side'] == 'BUY' else -row['traded_volume'] + 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 @@ -115,14 +115,14 @@ class TState: self.items[code] = TStateItem(code, -delta) for row in rows: - item = self.items.setdefault(row['code'], TStateItem(row['code'])) - self._reset(item, row['insert_date']) - qty, amount = row['traded_volume'], row['trade_amount'] - if row['local_order_id'].startswith('zt-base-'): + item = self.items.setdefault(row['stock_code'], TStateItem(row['stock_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['local_order_id'] + item.base_order_id = row['order_local_id'] else: self._apply_t_deal(item, row) self.deals.append(row) @@ -141,15 +141,10 @@ class TState: self._store.save( { code: { - 'code': item.code, - 'base_order_id': item.base_order_id, - 'base_qty': item.base_qty, - 'base_cost': item.base_cost, - 'added_order_id': item.added_order_id, - 'added_num': item.added_num, - 'added_qty': item.added_qty, - 'added_cost': item.added_cost, - 'status': item.phase, + '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() }, @@ -163,19 +158,21 @@ class TState: self._store.load() self.items = {} for code, position in self._store.positions.items(): - position = dict(position) - position['phase'] = position.pop('status') - self.items[code] = TStateItem(**position) + 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: - if not deal['local_order_id'].startswith('zt-base-') and deal['code'] in self.items: - self._apply_t_deal(self.items[deal['code']], deal) - for code, item in self.items.items(): - if self._store.positions[code]['status'] == READY: - item.phase, item.trade_date = READY, '' - item.sell_qty = item.buy_qty = 0 - item.sell_price = item.buy_cost = 0.0 + 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 81b0fbe..34d465b 100644 --- a/py-client/tests/test_deal_model.py +++ b/py-client/tests/test_deal_model.py @@ -1,75 +1,89 @@ -import sqlite3 +import ast import tempfile import unittest -from contextlib import closing -from dataclasses import asdict +from dataclasses import asdict, fields +from datetime import datetime 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 sdk.models import DealItem +from sdk.models import Assets, DealItem, OrderItem, PositionItem from sdk.portfolio import PortfolioMixin -class DealModelTests(unittest.TestCase): +class ApiModelTests(unittest.TestCase): def setUp(self): - self.raw = { - 'm_strOrderSysID': 'sys-123', - 'm_strInstrumentID': '600000', - 'm_strExchangeID': 'SH', - 'm_strInstrumentName': 'Test stock', - 'm_nOffsetFlag': '24', - 'm_nOrderStatus': '56', - 'm_nVolumeTotal': '0', - 'm_nVolumeTraded': '200', - 'm_nOrderTime': '101530', - 'm_strInsertDate': '20260907', - 'm_strInsertTime': '10:15:30', - 'm_strRemark': 'zt-t-sell-local1|zt', - 'm_dPrice': '15.8', - 'm_dTradePrice': '15.6', - 'm_dTradeAmount': '3120', - } + source = Path(__file__).resolve().parents[2] / 'api' / 'qmt_rest_new.py' + names = {'format_assets', 'format_holding', 'format_orders', 'format_deals'} + nodes = [n for n in ast.parse(source.read_text(encoding='utf-8')).body + if isinstance(n, ast.FunctionDef) and n.name in names] + ns = {'HTTPError': RuntimeError} + exec(compile(ast.Module(body=nodes, type_ignores=[]), str(source), 'exec'), ns) + attrs = {n.attr: '' if n.attr.startswith('m_str') else 0 + for node in nodes for n in ast.walk(node) + if isinstance(n, ast.Attribute) and n.attr.startswith('m_')} + attrs.update(m_strInstrumentID='600000', m_strExchangeID='SH', + m_strOrderSysID='sys1', m_strRemark='trend-BUY-1|trend', + m_nOffsetFlag=23, m_nOrderStatus=56, m_nVolume=100, + m_nVolumeTraded=100, m_nVolumeTotalOriginal=100, + m_dPrice=10.0, m_dTradeAmount=1000.0, m_dBalance=2000.0, + m_dAvailable=1000.0, m_strInsertDate='20260907', + m_strInsertTime='100000', m_strTradeDate='20260907', m_strTradeTime='100000') + obj = SimpleNamespace(**attrs) + self.assets = ns['format_assets']([obj]) + self.positions = ns['format_holding']([obj]) + self.orders = ns['format_orders']([obj]) + self.deals = ns['format_deals']([obj]) + self.client = PortfolioMixin() + self.client._get_json = { + '/api/portfolio/assets': self.assets, '/api/portfolio/positions': self.positions, + '/api/portfolio/order': self.orders, '/api/portfolio/deal': self.deals, + '/api/portfolio': {'assets': self.assets, 'positions': self.positions, 'orders': self.orders}, + }.__getitem__ - def test_all_api_fields_are_parsed(self): - self.assertEqual(asdict(DealItem.from_trade_detail(self.raw)), { - 'sys_order_id': 'sys-123', 'local_order_id': 'zt-t-sell-local1', - 'code': '600000.SH', 'instrument_id': '600000', 'exchange_id': 'SH', - 'name': 'Test stock', 'offset_flag': '24', 'side': 'SELL', 'status': '56', - 'remaining_volume': 0, 'traded_volume': 200, 'order_time': 101530, - 'insert_date': '20260907', 'insert_time': '10:15:30', - 'remark': 'zt-t-sell-local1|zt', 'price': 15.8, - 'trade_price': 15.6, 'trade_amount': 3120.0, - }) + def test_models_exactly_match_api_keys_and_values(self): + for model, row in ((Assets, self.assets), (PositionItem, self.positions['600000.SH']), + (OrderItem, self.orders[0]), (DealItem, self.deals[0])): + self.assertEqual({field.name for field in fields(model)}, set(row)) + self.assertEqual(asdict(model(**row)), row) - def test_api_deals_response_returns_deal_items(self): - client = PortfolioMixin() - for response in ({'deals': [self.raw]}, [self.raw]): - with self.subTest(response_type=type(response).__name__): - client._get_json = lambda path: response - deals = client.deals() - self.assertIsInstance(deals[0], DealItem) - self.assertEqual(deals[0].sys_order_id, 'sys-123') - self.assertEqual(deals[0].traded_volume, 200) + def test_all_endpoints(self): + self.assertEqual(asdict(self.client.assets()), self.assets) + codes, positions = self.client.positions() + self.assertEqual(codes, ['600000.SH']) + self.assertEqual(asdict(positions[0]), self.positions[codes[0]]) + self.assertEqual(asdict(self.client.orders()[0]), self.orders[0]) + self.assertEqual(asdict(self.client.deals()[0]), self.deals[0]) + portfolio = self.client.portfolio() + self.assertEqual(asdict(portfolio.positions[codes[0]]), self.positions[codes[0]]) + self.assertEqual(asdict(portfolio.orders[0]), self.orders[0]) - def test_model_matches_sql_columns_and_restart(self): - deal = DealItem.from_trade_detail(self.raw) + def test_derived_properties_and_order_cache(self): + order = self.client.orders()[0] + self.assertEqual(order.side, 'BUY') + self.assertEqual(order.local_order_id, 'trend-BUY-1') + self.assertEqual(order.created_at, datetime(2026, 9, 7, 10)) + order.order_status = 50 + order.insert_date = '20000101' + client = Mock() + book = ActiveOrders('trend') + book.refresh(client, [order]) + client.cancel_by_id.assert_called_once_with('sys1') + self.assertTrue(book.busy('600000.SH', 'BUY')) + + def test_storage_and_price_fallback(self): + deal = self.client.deals()[0] with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / 'state.db' book = OrderBook(path) book.sync_deals([deal]) - with closing(sqlite3.connect(path)) as db: - columns = {row[1] for row in db.execute('PRAGMA table_info(deals)')} - self.assertEqual(columns, {'id', *asdict(deal)}) - loaded = OrderBook(path).deals['sys-123'] - expected = asdict(deal) - expected['insert_date'] = '2026-09-07' - self.assertEqual({key: value for key, value in loaded.items() if key != 'id'}, expected) - - def test_missing_amount_uses_execution_price_not_order_price(self): - deal = DealItem.from_trade_detail({**self.raw, 'm_dTradeAmount': '0'}) - row = OrderBook.deal_record(deal) - self.assertEqual(row['trade_amount'], 3120) - deal.trade_price = 0 + loaded = OrderBook(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) diff --git a/py-client/tests/test_orderbook.py b/py-client/tests/test_orderbook.py index 7b7a027..9c644a4 100644 --- a/py-client/tests/test_orderbook.py +++ b/py-client/tests/test_orderbook.py @@ -2,6 +2,7 @@ import sqlite3 import tempfile import unittest from contextlib import closing +from dataclasses import asdict, fields from pathlib import Path from datetime import datetime from unittest.mock import patch @@ -23,18 +24,12 @@ class OrderBookTests(unittest.TestCase): def deal(self, kind, sys_order_id, qty, price, date='2026-09-01'): prefix = {'base': 'zt-base-', 'sell': 'zt-t-sell-', 'buy': 'zt-t-buy-'}[kind] - return DealItem.from_trade_detail({ - 'm_strOrderSysID': sys_order_id, - 'm_strInstrumentID': '600000', 'm_strExchangeID': 'SH', - 'm_strInstrumentName': 'Test stock', - 'm_nOffsetFlag': '24' if kind == 'sell' else '23', - 'm_nOrderStatus': '56', 'm_nVolumeTotal': '0', - 'm_nVolumeTraded': str(qty), 'm_nOrderTime': '100000', - 'm_strInsertDate': date, 'm_strInsertTime': '10:00:00', - 'm_strRemark': prefix + 'order1|zt', - 'm_dPrice': str(price + 1), 'm_dTradePrice': str(price), - 'm_dTradeAmount': str(qty * price), - }) + return DealItem( + order_sys_id=sys_order_id, stock_code='600000.SH', + offset_flag=24 if kind == 'sell' else 23, + volume=qty, price=price, trade_amount=qty * price, + trade_date=date, trade_time='10:00:00', remark=prefix + 'order1|zt', + ) def test_partial_fills_restart_dedup_and_daily_cycle(self): state = TState(self.path) @@ -83,10 +78,12 @@ class OrderBookTests(unittest.TestCase): insert.assert_called_once() self.assertEqual(len(insert.call_args.args[1]), 2) self.assertEqual(book.deals_sys_ids, {'d1', 'd2'}) - self.assertEqual(book.deals['d1']['insert_date'], '2026-09-01') - self.assertEqual(book.deals['d2']['traded_volume'], 60) + 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) self.assertEqual(book.deals_sys_ids, {'d1', 'd2'}) + self.assertEqual(book.deals['d1']['order_local_id'], 'zt-base-order1') with patch.object(book, '_insert_deals') as insert: book.sync_deals([first, second]) book.sync_deals([]) @@ -97,25 +94,25 @@ class OrderBookTests(unittest.TestCase): book = OrderBook(self.path) first = self.deal('base', 'd1', 100, 10) invalid = self.deal('base', 'd2', 100, 10) - invalid.side = 'INVALID' + invalid.offset_flag = -1 with self.assertRaises(sqlite3.IntegrityError): book.sync_deals([first, invalid]) self.assertEqual(book.deals, {}) self.assertEqual(book.deals_sys_ids, set()) self.assertEqual(OrderBook(self.path).deals, {}) - invalid.side = 'BUY' + 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.save({'600000.SH': {'code': '600000.SH', 'status': READY}}, []) + writer.sync_positions([PositionItem(stock_code='600000.SH', volume=100)]) writer.sync_deals([self.deal('base', 'd1', 100, 10)]) book.load() - self.assertEqual(book.positions['600000.SH']['status'], READY) + self.assertEqual(book.positions['600000.SH']['volume'], 100) self.assertEqual(book.deals_sys_ids, {'d1'}) - self.assertEqual(book.deals['d1']['local_order_id'], 'zt-base-order1') + self.assertEqual(book.deals['d1']['remark'], 'zt-base-order1|zt') def test_first_start_after_full_sale_keeps_buyback_quantity(self): state = TState(self.path) @@ -148,12 +145,17 @@ class OrderBookTests(unittest.TestCase): tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} self.assertEqual(tables, {'positions', 'deals', 'sqlite_sequence'}) columns = {row[1] for row in db.execute('PRAGMA table_info(deals)')} - self.assertFalse({'kind', 'confirmed_date', 'deal_ids', 'deal_id', 'order_id', 'qty', 'filled_qty', 'filled_cost', 'amount', 'trade_date', 'trade_time'} & columns) - self.assertTrue({'sys_order_id', 'local_order_id'} <= columns) + self.assertEqual(columns, {'id', 'order_local_id', *(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_code', 'idx_positions_base_order_id', - 'idx_positions_added_order_id', 'idx_deals_local_order_id', - 'idx_deals_code_date', 'idx_deals_date_time'} <= indexes) + self.assertTrue({'idx_positions_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']), + ('idx_deals_order_ref', ['order_local_id']), + ('idx_deals_stock_code_date', ['stock_code']), + ('idx_deals_date_time', ['trade_date']), + ): + self.assertEqual([row[2] for row in db.execute(f'PRAGMA index_info({index})')], expected) self.assertNotIn('kind', state.deals[0]) self.assertEqual(TState(self.path).deals, state.deals) state.deals.append(dict(state.deals[0])) @@ -168,29 +170,23 @@ class OrderBookTests(unittest.TestCase): def test_position_columns_defaults_indexes_and_stable_id(self): store = OrderBook(self.path) with closing(sqlite3.connect(self.path)) as db: - columns = [row[1] for row in db.execute('PRAGMA table_info(positions)')] - self.assertEqual(columns, [ - 'id', 'code', 'base_order_id', 'base_qty', 'base_cost', - 'added_order_id', 'added_num', 'added_qty', 'added_cost', 'status', - ]) + 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_code', 'idx_positions_base_order_id', 'idx_positions_added_order_id', - }) - store.save({'600000.SH': {'code': '600000.SH', 'status': READY}}, []) - position = store.positions['600000.SH'] - first_id = position['id'] - self.assertGreater(first_id, 0) - self.assertEqual(position['base_order_id'], '') - self.assertEqual(position['base_qty'], 0) - self.assertEqual(position['added_cost'], 0.0) - position.update(base_order_id='base1', base_qty=200, base_cost=10.5, - added_order_id='add1', added_num=1, added_qty=100, added_cost=9.0, - status='ACTIVE') - store.save({'600000.SH': position}, []) - self.assertEqual(store.positions['600000.SH'], position) - store.save({}, []) - store.save({'600001.SH': {'code': '600001.SH', 'status': READY}}, []) + self.assertEqual(indexes, {'idx_positions_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] + first_id = saved['id'] + self.assertEqual({k: v for k, v in saved.items() if k != 'id'}, asdict(position)) + 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, {}) + store.sync_positions([PositionItem(stock_code='600001.SH')]) self.assertGreater(store.positions['600001.SH']['id'], first_id) def test_base_split_fills_and_snapshot_do_not_double_count(self): @@ -207,12 +203,12 @@ class OrderBookTests(unittest.TestCase): state = TState(self.path) first = self.deal('base', 'd1', 100, 10, '20260901') other = self.deal('base', 'd2', 100, 10) - other.local_order_id = 'trend-base-order' + other.remark = 'trend-base-order' state.reconcile([], [first, other]) - first.insert_date = '2026-09-01' + first.trade_date = '2026-09-01' state.reconcile([], [first]) self.assertEqual(len(state.deals), 1) - self.assertEqual(state.deals[0]['insert_date'], '2026-09-01') + self.assertEqual(state.deals[0]['trade_date'], '2026-09-01') if __name__ == '__main__': diff --git a/py-client/tests/test_trend_collector.py b/py-client/tests/test_trend_collector.py new file mode 100644 index 0000000..51b1d0d --- /dev/null +++ b/py-client/tests/test_trend_collector.py @@ -0,0 +1,88 @@ +import importlib +import io +import logging +import unittest +from concurrent.futures import Future +from contextlib import ExitStack, redirect_stdout +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from libs import collector +from sdk import Assets, PositionItem +from strategy.trend import boot + + +class TrendCollectorTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + with patch('logging.FileHandler', return_value=logging.NullHandler()): + cls.app = importlib.import_module('main') + + def setUp(self): + old_snapshot = boot._collector_snapshot + self.addCleanup(setattr, boot, '_collector_snapshot', old_snapshot) + boot._collector_snapshot = None + + def test_submission_reads_latest_cache_and_skips_empty(self): + with patch.object(collector, 'collector_push') as push: + collector.submit_trend_data() + push.assert_not_called() + boot._cache_portfolio('account', Assets(available=100), []) + assets = Assets(available=200) + positions = [PositionItem(stock_code='600000.SH', volume=100)] + boot._cache_portfolio('account', assets, positions) + collector.submit_trend_data() + push.assert_called_once_with('account', assets, positions) + uploaded = push.call_args.args + uploaded[1].available = 0 + uploaded[2].clear() + self.assertEqual(boot.get_collector_snapshot()[1].available, 200) + self.assertEqual(len(boot.get_collector_snapshot()[2]), 1) + + def test_run_once_caches_portfolio_without_submitting_data(self): + completed = Future() + completed.set_result(None) + run = SimpleNamespace( + client=Mock(), orders=Mock(), executor=Mock(), + account_cfg=SimpleNamespace(account_id='account', min_cash_ratio=0.1), + ) + assets = Assets(available=100, total=1000) + run.client.portfolio.return_value = SimpleNamespace(assets=assets, positions={}, orders=[]) + run.client.full_tick.return_value = {} + run.executor.submit.return_value = completed + with patch.object(boot, 'trading_time', return_value=True), \ + patch.object(boot, 'market_allow_open', return_value=True), \ + patch.object(collector, 'collector_push') as push, redirect_stdout(io.StringIO()): + boot.RunOnce(run, []) + self.assertEqual(boot.get_collector_snapshot(), ('account', assets, [])) + push.assert_not_called() + run.executor.submit.assert_called_once_with(boot.manage_positions, run, {}, [], True, 100) + + def test_main_registers_five_minute_collector_job(self): + for strategy in ('trend', 'zt'): + with self.subTest(strategy=strategy), ExitStack() as stack: + scheduler = Mock(running=True) + stack.enter_context(patch.object(self.app, 'BackgroundScheduler', return_value=scheduler)) + stack.enter_context(patch.object(self.app, 'require_windows', return_value=True)) + stack.enter_context(patch.object(self.app, 'check_single_instance', return_value=True)) + stack.enter_context(patch.object(self.app, 'wait_for_qmt_api')) + stack.enter_context(patch.object(self.app.config, 'load')) + stack.enter_context(patch.object(self.app.config, 'global_config', SimpleNamespace(api_host='unused'))) + stack.enter_context(patch.object(self.app.config, 'account_config', SimpleNamespace(strategy=strategy))) + stack.enter_context(patch.dict(self.app.STRATEGIES, { + strategy: SimpleNamespace(start_strategy=Mock()), + })) + self.assertEqual(self.app.main(), 0) + jobs = [call for call in scheduler.add_job.call_args_list + if call.kwargs.get('id') == 'trend_collector'] + self.assertEqual(len(jobs), 1) + if jobs: + self.assertIs(jobs[0].args[0], collector.submit_trend_data) + self.assertEqual(jobs[0].kwargs['trigger'], 'interval') + self.assertEqual(jobs[0].kwargs['minutes'], 5) + scheduler.start.assert_called_once() + scheduler.shutdown.assert_called_once_with(wait=True) + + +if __name__ == '__main__': + unittest.main()