diff --git a/py-client/etc/yin_fei.yaml b/py-client/etc/yin_fei.yaml index 01b1074..cccfef2 100644 --- a/py-client/etc/yin_fei.yaml +++ b/py-client/etc/yin_fei.yaml @@ -1,6 +1,6 @@ account_id: 8891110937 host_key: yin_fei -buy_value: 5000 +buy_value: 10000 min_cash_ratio: 0.10 loss_trigger_pct: -10 grid_step_pct: 1 diff --git a/py-client/libs/state.py b/py-client/libs/state.py index 136b5cf..5c29e1e 100644 --- a/py-client/libs/state.py +++ b/py-client/libs/state.py @@ -1,7 +1,6 @@ """SQLite 策略状态与成交存储；每个数据库仅使用一个写入者，不做数据迁移。""" import math -import json import logging as log import sqlite3 from contextlib import closing @@ -29,12 +28,6 @@ CREATE TABLE IF NOT EXISTS state ( -- 每个证券仅保留一条策略状态。 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, @@ -104,8 +97,8 @@ class State: self.deals = deals self.deals_sys_ids = set(deals) - def sync_deals(self, deals: list[DealItem]) -> None: - """按系统成交编号去重，整批写入成功后刷新缓存。""" + def sync_deals(self, deals: list[DealItem], *, archived: bool = False) -> 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: @@ -127,62 +120,44 @@ class State: 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', + '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), + amount, date, deal.trade_time, deal.remark, deal.close_profit, int(archived)), ) self.load() - def archiving(self) -> dict[str, str]: - """按证券从持仓基准重放成交；失败证券保留未归档记录并返回原因。""" + def archiving(self, *, base_order_prefix: str = '') -> 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)' + 'SELECT DISTINCT stock_code FROM deals WHERE is_arch = 0 AND offset_flag IN (23, 24, 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 + state = dict(current) if current else asdict(StateItem(stock_code=code)) deals = db.execute( - 'SELECT * FROM deals WHERE stock_code = ? AND offset_flag IN (48, 49) ' + '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'] == 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'] + if deal['offset_flag'] in (23, 48): + bucket = 'base' if base_order_prefix and deal['order_local_id'].startswith(base_order_prefix) 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['added_qty'] = total - state['added_order_local_id'] = deal['order_local_id'] - state['added_created_at'] = f"{deal['trade_date']} {deal['trade_time']}".strip() + 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: - # 卖出之后的买入属于新交易，不再与初始快照匹配。 - covered = True total = state['base_qty'] + state['added_qty'] if qty > total: raise ValueError(f'Sell volume {qty} exceeds recorded holdings {total}') @@ -198,7 +173,7 @@ class State: 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) @@ -211,7 +186,7 @@ class State: ) db.execute( 'UPDATE deals SET is_arch = 1 WHERE stock_code = ? ' - 'AND is_arch = 0 AND offset_flag IN (48, 49)', (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') @@ -222,28 +197,24 @@ class State: self.load() return errors - def sync_state(self, positions: list[PositionItem]) -> None: + def sync_state(self, positions: list[PositionItem], *, remove_missing: bool = True) -> None: """同步完整持仓：无状态则插入底仓，已有则保留，清仓则删除。 数量为零或未出现在完整持仓列表中的证券视为已清仓；空列表清空状态。 + 底仓已包含的历史成交不应再次归档；后续成交须先归档，再同步持仓。 + remove_missing=False 时仅接纳新底仓，减仓由成交归档处理。 """ # 传入完整账户持仓；同步时间作为新增底仓的创建时间。 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), + existing = {row['stock_code'] for row in db.execute('SELECT stock_code FROM state')} + if remove_missing: + db.executemany( + 'DELETE FROM state WHERE stock_code = ?', + [(code,) for code in existing if code not in holdings], ) - 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 diff --git a/py-client/main.py b/py-client/main.py index 82ff8f5..7c85ad5 100644 --- a/py-client/main.py +++ b/py-client/main.py @@ -36,6 +36,7 @@ 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) @@ -46,6 +47,7 @@ class StrategyDefinition: STRATEGIES = { "trend": StrategyDefinition("Trend", StartTrend), + "zt": StrategyDefinition("ZT", StartZT), } def require_windows() -> bool: diff --git a/py-client/strategy/zt/boot.py b/py-client/strategy/zt/boot.py index 1c0b782..f9d5c66 100644 --- a/py-client/strategy/zt/boot.py +++ b/py-client/strategy/zt/boot.py @@ -1,196 +1,96 @@ -"""做 T 策略启动器。 +"""ZT 启动与串行调度：成交同步、买回、卖出、建仓。""" -该模块负责组合 SDK、配置、状态存储和做 T 策略组件，供 main.py 调用。 -""" - -from concurrent.futures import Future, ThreadPoolExecutor import logging as log import time -from datetime import datetime, time as clock_time +from datetime import datetime from pathlib import Path import config from libs.calc import trading_time -from libs.market import market_allow_open -from libs.signal import SignalItem, init_signals from libs.collector import collector_push from libs.grid_take_profit import GridTrailingTracker -from sdk import Client +from libs.market import market_allow_open +from libs.order import OrderBook from libs.overview import Overview -from libs.order import BUSY_STATUSES, OrderBook -from libs.watch import DipWatch from libs.runtime import Runtime +from libs.signal import SignalItem, init_signals +from libs.state import State +from libs.watch import DipWatch +from sdk import Client, DealItem, PositionItem from .open import open_signal -from .positions import manage_positions +from .positions import manage_positions, t_rounds def StartZT() -> None: - """初始化做 T 策略，并以 30 秒间隔持续执行。""" - with Client( - config.global_config.qmt_base_url, - 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" - # ) - executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="zt") + with Client(config.global_config.qmt_base_url, config.global_config.qmt_token, config.HTTP_TIMEOUT) as client: + state = State(Path(config.global_config.qmt_data_dir) / f'zt_{config.account_config.account_id}_state.db') run = Runtime( - client=client, - global_cfg=config.global_config, - account_cfg=config.account_config, - orders=OrderBook("zt"), - open_watch=DipWatch(), - add_watch=DipWatch(), + client=client, global_cfg=config.global_config, account_cfg=config.account_config, + orders=OrderBook('zt'), open_watch=DipWatch(), add_watch=DipWatch(), profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct), - executor=executor ) - + # 先读取成交，再读取持仓，减少成交已入账而快照仍未更新的情况。 + deals = client.deals() portfolio = client.portfolio() - assets = portfolio.assets positions = list(portfolio.positions.values()) + sync_account_state(state, positions, deals, initialize=not state.state and not state.deals) run.orders.refresh(client, portfolio.orders) - - # 获取本策略的信号开仓数据 - signals = init_signals(config.global_config,["dcm"]) - log.info("[启动] ZT 策略已启动，账户=%s，信号=%d，持仓=%d", - config.account_config.account_id, - len(signals), - len(positions), - ) - - Overview(assets, positions, config.account_config) - - DEFAULT_TICK_INTERVAL = 30 - while True: - lt = time.localtime() - if (lt.tm_hour, lt.tm_min, lt.tm_sec) >= (15, 0, 0): - log.info("[Trend] 已到 15:00，结束趋势策略") - return - current_sec = lt.tm_sec - - # 计算距离下一个目标时间点（0秒或30秒）的等待时间 - if current_sec < DEFAULT_TICK_INTERVAL: - wait_seconds = DEFAULT_TICK_INTERVAL - current_sec - elif current_sec < 60: - wait_seconds = 60 - current_sec - else: - wait_seconds = DEFAULT_TICK_INTERVAL - - # 等待到目标时间点 - time.sleep(wait_seconds) - - # 单轮失败不能杀死唯一的交易定时线程。 + signals = init_signals(config.global_config, ['dcm']) + Overview(portfolio.assets, positions, config.account_config) + log.info('[ZT] 启动，账户=%s，信号=%d', config.account_config.account_id, len(signals)) + while datetime.now().hour < 15: try: RunOnce(run, state, signals) - except Exception as e: - log.error( - f"[Trend] 本 tick 执行失败，下一 tick 继续: {e}", exc_info=True - ) + except Exception: + log.exception('[ZT] 本轮失败，下一轮重试') + time.sleep(30 - time.time() % 30) + # 收盘后补记最后一轮成交，不再下单。 + sync_account_state(state, list(client.portfolio().positions.values()), client.deals()) -def RunOnce(run: Runtime, state: TState, signals: list[SignalItem]) -> None: - """账户快照 → 成交对账 → 做 T 管理 → dcm 建仓，共用一份资金预算。""" + +def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None: now = datetime.now() if not trading_time(now): return - - started_at = time.monotonic() - - # 1. 一次获取资产、持仓和订单，并清理过期订单。 - try: - portfolio = run.client.portfolio() - assets = portfolio.assets - deals = run.client.deals() - position_codes = list(portfolio.positions) - positions = list(portfolio.positions.values()) - run.orders.refresh(run.client, portfolio.orders) - state.reconcile(positions,deals) - except Exception: - log.exception("[Portfolio] 刷新账户快照失败") + deals = run.client.deals() + portfolio = run.client.portfolio() + assets = portfolio.assets + positions = list(portfolio.positions.values()) + run.orders.refresh(run.client, portfolio.orders) + sync_account_state(state, positions, deals) + # 收盘集合竞价前停止提交新委托，继续保存成交。 + if (now.hour, now.minute) >= (14, 57): return - - futures: list[tuple[str, Future]] = [ - ( - "数据提交", - run.executor.submit( - collector_push, - run.account_cfg.account_id, - assets, - positions, - ), - ) - ] - - # 2. 验证可用资金；低于资金安全线时禁止开新仓。 - allow_open_by_cash = ( - assets.available >= assets.total * run.account_cfg.min_cash_ratio - ) - if not allow_open_by_cash: - log.info( - "[Status] 禁止开仓：可用资金不足，可用=%.2f，总资产=%.2f", - assets.available, - assets.total, - ) - - # 3. 获取大盘状态，只有大盘信号允许时才执行开仓。 - market_ok = market_allow_open() - - # 4. 验证有效开仓信号：排除已有持仓和未决订单。 - allow_open: list[SignalItem] = [] - allow_codes: list[str] = [] - for signal in signals: - if signal.code not in portfolio.positions: - allow_open.append(signal) - allow_codes.append(signal.code) - - if allow_open and not market_ok: - log.info("[开仓] 禁止开仓：大盘信号不允许，候选=%d", len(allow_open)) - - # 5. 获取持仓和待开仓证券的实时行情 tick。 - all_codes = list(dict.fromkeys(position_codes + allow_codes)) + rounds = t_rounds(state) + pending = {code for code, item in rounds.items() if item['sold'] > item['bought']} + candidates = {s.code: s for s in signals if s.code not in portfolio.positions + and s.code not in state.state and s.code not in pending} + codes = list(dict.fromkeys(list(state.state) + sorted(pending) + list(candidates))) + ticks = run.client.full_tick(codes) if codes else {} + force = (now.hour, now.minute) >= (14, 50) + available = manage_positions(run, state, ticks, positions, rounds, assets.available, now.date().isoformat(), force) + # 尚未买回时不分走资金；买回与新建仓使用同一份剩余资金。 + if not force and not pending and available >= assets.total * run.account_cfg.min_cash_ratio: + if candidates and market_allow_open(): + budget = max(0.0, available - assets.total * run.account_cfg.min_cash_ratio) + open_signal(run, ticks, list(candidates.values()), budget) try: - ticks = run.client.full_tick(all_codes) + collector_push(run.account_cfg.account_id, assets, positions) except Exception: - log.exception("[行情] 获取行情失败，代码数量=%d", len(all_codes)) + log.exception('[ZT] 持仓上报失败') + + +def sync_account_state( + state: State, positions: list[PositionItem], deals: list[DealItem], *, initialize: bool = False, +) -> None: + """初次持仓作为底仓；后续只按成交减仓，避免延迟快照删除持仓。""" + zt_deals = [d for d in deals if d.get_local_order_id.startswith('zt-')] + state.sync_deals(zt_deals, archived=initialize) + if initialize: + state.sync_state(positions) return - - log.info( - "[RunOnce] 本轮就绪，持仓=%d，候选=%d，大盘允许=%s，资金允许=%s", - len(positions), - len(allow_open), - market_ok, - allow_open_by_cash, - ) - - # 启动线程，开始计算 - # 7. 持仓计算。 - futures.append( - ( - "持仓计算", - run.executor.submit( - manage_positions, run, ticks, positions, market_ok, assets.available - ), - ) - ) - - # 8. 开仓计算：必须同时存在有效信号且大盘允许开仓。 - if allow_open and market_ok and allow_open_by_cash: - futures.append( - ("开仓计算", run.executor.submit(open_signal, run, ticks, allow_open)) - ) - - # 9. 开始执行 - for name, future in futures: - _wait_worker(name, future) - log.info( - "[RunOnce] 本轮完成，耗时=%d毫秒", int((time.monotonic() - started_at) * 1000) - ) - - -def _wait_worker(name: str, future: Future) -> None: - """保留单轮继续运行的语义，分别记录工作线程异常。""" - try: - future.result() - except Exception: - log.exception("[运行] %s线程失败", name) + errors = state.archiving(base_order_prefix='zt-base-') + if errors: + raise ValueError(f'ZT 成交归档失败：{errors}') + traded = {d['stock_code'] for d in state.deals.values()} + state.sync_state([p for p in positions if p.stock_code not in traded], remove_missing=False) diff --git a/py-client/strategy/zt/open.py b/py-client/strategy/zt/open.py index 46584dc..ae7ca3c 100644 --- a/py-client/strategy/zt/open.py +++ b/py-client/strategy/zt/open.py @@ -4,7 +4,6 @@ from datetime import datetime import logging as log import math -from libs.calc import calc_buy_volume from sdk import OP_BUY from libs.runtime import Runtime from libs.order import PlaceOrderRequest @@ -12,11 +11,11 @@ from libs.order import PlaceOrderRequest def open_signal(run: Runtime, ticks, signals, available: float) -> float: """逐个验证开仓信号并提交买入委托，返回本轮剩余资金。""" + now = datetime.now() + if (now.hour, now.minute) >= (14, 50): + return available for item in signals: try: - now = datetime.now() - if (now.hour, now.minute) >= (14, 50): - break if item.code in run.account_cfg.excluded_codes: continue # 由委托簿检查活动委托，防止重复下单。 @@ -36,16 +35,16 @@ def open_signal(run: Runtime, ticks, signals, available: float) -> float: continue # 根据单笔买入金额计算整手数量，并预留少量价差和费用。 budget = min(run.account_cfg.buy_value, available) - volume = calc_buy_volume(price, budget) + volume = int(budget / (price * 1.01)) // 100 * 100 amount = price * volume * 1.01 - if volume <= 0 or price * volume > budget or amount > available: + if volume < (200 if item.code.startswith('688') else 100): continue # 等待价格从观察低点反弹，防止直接接下跌中的“飞刀”。 if not run.open_watch.triggered("ZT 建仓", item.code, price): continue order_id = run.orders.new_order_id("base") request = PlaceOrderRequest( - OP_BUY, item.code, volume, order_id, "zt", kind="base" + OP_BUY, item.code, volume, order_id, "zt" ) # 即使响应丢失，本轮也预留资金；状态簿只在取得实际成交后入账。 available -= amount diff --git a/py-client/strategy/zt/positions.py b/py-client/strategy/zt/positions.py index 72dd200..17cf87b 100644 --- a/py-client/strategy/zt/positions.py +++ b/py-client/strategy/zt/positions.py @@ -1,115 +1,101 @@ -"""日内先卖后买的做 T 规则，不包含趋势补仓或整仓止盈。""" +"""先卖后买做 T；持仓存在 State，未买回数量从实际成交计算。""" import logging as log import math 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 libs.state import State +from sdk import OP_BUY, OP_SELL, PositionItem + + +def t_rounds(store: State) -> dict[str, dict]: + """每只证券保留最近一轮 T，跨日未买回的数量继续保留。""" + rounds = {} + for deal in sorted(store.deals.values(), key=lambda d: ( + d['trade_date'], d['trade_time'].replace(':', ''), d['id'] + )): + order = deal['order_local_id'] + code = deal['stock_code'] + if order.startswith('zt-t-sell-') and deal['offset_flag'] in (24, 49): + item = rounds.get(code) + if item is None or item['bought'] >= item['sold']: + item = rounds[code] = dict(sold=0, bought=0, amount=0.0, date='') + item['sold'] += deal['volume'] + item['amount'] += deal['trade_amount'] + item['date'] = deal['trade_date'] + elif order.startswith('zt-t-buy-') and deal['offset_flag'] in (23, 48) and code in rounds: + rounds[code]['bought'] += deal['volume'] + rounds[code]['date'] = deal['trade_date'] + return rounds def manage_positions( - run: Runtime, - ticks, - positions: list[PositionItem], - available: float, - today: str, - force_buy_back: bool = False, + run: Runtime, store: State, ticks, positions: list[PositionItem], + rounds: dict[str, dict], available: float, today: str, force_buy_back: bool = False, ) -> float: - """遍历本地底仓记录；全部卖出后即使持仓快照为空，也必须处理买回。""" - by_code = {position.stock_code: position for position in positions} - for code, state in list(state_store.items.items()): + """先偿还买回欠仓；同一证券当天完成一轮后不再卖出。""" + by_code = {p.stock_code: p for p in positions} + codes = dict.fromkeys(list(rounds) + list(store.state)) + for code in sorted(codes, key=lambda c: not (c in rounds and rounds[c]['sold'] > rounds[c]['bought'])): try: if code in run.account_cfg.excluded_codes: continue - if run.orders.busy(code, "BUY") or run.orders.busy(code, "SELL"): + if run.orders.busy(code, 'BUY') or run.orders.busy(code, 'SELL'): continue tick = ticks.get(code) price = tick.last_price if tick else 0.0 if not math.isfinite(price) or price <= 0: continue + item = rounds.get(code) + row = store.state.get(code, {}) position = by_code.get(code) - actual_qty = position.volume if position else 0 - expected_qty = state.base_qty - state.sell_qty + state.buy_qty - # 快照延迟或手动增减仓不能当作新的做 T 信号，先核对数量差异。 - if actual_qty != expected_qty: - log.warning( - "[ZT 持仓] %s 数量不符，记录=%d，实际=%d，暂停交易", - code, - expected_qty, - actual_qty, - ) + recorded = row.get('base_qty', 0) + row.get('added_qty', 0) + if recorded != (position.volume if position else 0): + log.warning('[ZT] %s 持仓快照与成交未对齐，等待下一轮', code) continue - if state.phase == SOLD: - available = _try_buy_back( - run, state, price, available, force_buy_back - ) - elif state.phase == READY and position and not force_buy_back: - if price <= run.account_cfg.zt_max_price: - _try_sell(run, state, position, price, today) + if item and item['sold'] > item['bought']: + volume = item['sold'] - item['bought'] + # 部分成交后的零股欠仓不能按普通买入申报，不扩大买回数量。 + minimum = 200 if code.startswith('688') else 100 + if not code.startswith('688'): + volume = volume // 100 * 100 + if volume < minimum: + log.warning('[ZT] %s 剩余买回 %d 股不满足申报数量，保留欠仓', code, item['sold'] - item['bought']) + continue + target = item['amount'] / item['sold'] * (1 - run.account_cfg.zt_buy_fall_pct / 100) + if not force_buy_back and price > target: + continue + amount = price * volume * 1.01 + if amount > available: + log.warning('[ZT] %s 买回资金不足，需要 %.2f，可用 %.2f', code, amount, available) + continue + if not force_buy_back and not run.add_watch.triggered('ZT 买回', code, price): + continue + available -= amount + request = PlaceOrderRequest(OP_BUY, code, volume, run.orders.new_order_id('t-buy'), 'zt') + if run.orders.place(run.client, request): + run.add_watch.forget(code) + log.info('[ZT 买回] %s %d 股%s', code, volume, '，尾盘买回' if force_buy_back else '') + continue + if force_buy_back or (item and item['date'] >= today) or not position or recorded <= 0: + continue + if price > run.account_cfg.zt_max_price: + continue + cost = (row['base_qty'] * row['base_price'] + row['added_qty'] * row['added_price']) / recorded + if cost <= 0: + continue + pnl = (price / cost - 1) * 100 + observation = run.profit_tracker.observe(f'{run.account_cfg.account_id}:{code}:{today}', pnl) + if observation.state != GridState.RETREAT or pnl < run.account_cfg.min_profit_pct: + continue + volume = int(min(position.can_use_volume, recorded * run.account_cfg.zt_sell_ratio)) // 100 * 100 + if volume < (200 if code.startswith('688') else 100): + continue + request = PlaceOrderRequest(OP_SELL, code, volume, run.orders.new_order_id('t-sell'), 'zt') + if run.orders.place(run.client, request): + log.info('[ZT 卖出] %s %d 股，按实际成交买回', code, volume) except Exception: - log.exception("[ZT 持仓] %s 处理异常，继续后续证券", code) - return available - - -def _try_sell( - run: Runtime, - state, - position: PositionItem, - price: float, - today: str, -) -> None: - """基于独立保存的底仓成本，用跨轮最高盈利网格判断做 T 卖出。""" - if state.base_cost <= 0: - return - pnl_rate = (price - state.base_cost) / state.base_cost * 100 - key = f"{run.account_cfg.account_id}:{state.code}:{today}" - observation = run.profit_tracker.observe(key, pnl_rate) - if observation.state != GridState.RETREAT: - return - volume = min( - position.can_use_volume, int(state.base_qty * run.account_cfg.zt_sell_ratio) - ) - volume = volume // 100 * 100 - if volume <= 0: - return - order_id = run.orders.new_order_id("t-sell") - request = PlaceOrderRequest( - OP_SELL, state.code, volume, order_id, "zt", kind="sell" - ) - if run.orders.place(run.client, request): - log.info("[ZT 卖出] %s %d 股，等待成交后确定买回数量和价格", state.code, volume) - - -def _try_buy_back( - run: Runtime, - state, - price: float, - available: float, - force: bool, -) -> float: - """按实际卖出均价下跌后反弹买回；尾盘不再受下跌幅度、反弹及价格上限限制。""" - target = state.sell_price * (1 - run.account_cfg.zt_buy_fall_pct / 100) - if not force and (price > target or price > run.account_cfg.zt_max_price): - return available - volume = state.sell_qty - state.buy_qty - amount = price * volume * 1.01 - if volume <= 0 or amount > available: - log.warning("[ZT 买回] %s 买回资金不足或数量无效，保留未完成轮次", state.code) - return available - if not force and not run.add_watch.triggered("ZT 买回", state.code, price): - return available - order_id = run.orders.new_order_id("t-buy") - request = PlaceOrderRequest(OP_BUY, state.code, volume, order_id, "zt", kind="buy") - # 本轮预留资金；状态簿只在取得实际成交后入账。 - available -= amount - if run.orders.place(run.client, request): - run.add_watch.forget(state.code) - log.info( - "[ZT 买回] %s %d 股，%s", - state.code, - volume, - "尾盘强制买回" if force else "下跌后反弹", - ) + log.exception('[ZT 持仓] %s 处理失败', code) return available diff --git a/py-client/tests/test_orderbook.py b/py-client/tests/test_orderbook.py index f90e3cc..054b563 100644 --- a/py-client/tests/test_orderbook.py +++ b/py-client/tests/test_orderbook.py @@ -9,15 +9,10 @@ from unittest.mock import patch from libs.state import State, StateItem from sdk import DealItem, PositionItem -from strategy.zt.state import DONE, READY, SOLD, TState class OrderBookTests(unittest.TestCase): def setUp(self): - clock = patch('strategy.zt.state.datetime') - self.clock = clock.start() - self.addCleanup(clock.stop) - self.clock.now.return_value = datetime(2026, 9, 1) self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.path = Path(self.tmp.name) / 'state.db' @@ -31,46 +26,18 @@ class OrderBookTests(unittest.TestCase): 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) - state.reconcile([PositionItem(stock_code='600000.SH', volume=200, open_price=10)], []) - self.assertEqual(state.deals, []) - first = self.deal('sell', 'd1', 40, 12) - second = self.deal('sell', 'd2', 60, 13) - state.reconcile([], [first]) - state = TState(self.path) - self.assertEqual(state.items['600000.SH'].phase, SOLD) - self.assertEqual(state.items['600000.SH'].sell_qty, 40) - state.reconcile([], [first, first, second]) - self.assertEqual(len(state.deals), 2) - self.assertAlmostEqual(state.items['600000.SH'].sell_price, 12.6) - self.clock.now.return_value = datetime.fromisoformat('2026-09-02') - state.reconcile([], [first, second]) - self.assertEqual(len(state.deals), 2) - self.assertEqual(state.items['600000.SH'].phase, SOLD) - b1 = self.deal('buy', 'd3', 40, 11, '2026-09-02') - b2 = self.deal('buy', 'd4', 60, 10, '2026-09-02') - state.reconcile([], [b1]) - self.assertEqual(state.items['600000.SH'].phase, SOLD) - state.reconcile([], [b1, b2]) - self.assertEqual(TState(self.path).items['600000.SH'].phase, DONE) - self.assertAlmostEqual(state.items['600000.SH'].buy_cost, 10.4) - self.clock.now.return_value = datetime.fromisoformat('2026-09-03') - state.reconcile([], []) - item = TState(self.path).items['600000.SH'] - self.assertEqual((item.phase, item.base_qty, item.base_cost, item.sell_qty), (READY, 200, 10, 0)) def test_json_is_never_read(self): legacy = self.path.with_suffix('.json') legacy.write_text('invalid JSON', encoding='utf-8') book = State(self.path) self.assertIsNone(book.load()) - self.assertEqual((book.items, book.deals, book.deals_sys_ids), ({}, {}, set())) + self.assertEqual((book.state, 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 = State(self.path) - self.assertEqual((book.items, book.deals, book.deals_sys_ids), ({}, {}, set())) + self.assertEqual((book.state, 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]) @@ -107,62 +74,12 @@ class OrderBookTests(unittest.TestCase): writer.sync_state([PositionItem(stock_code='600000.SH', volume=100)]) writer.sync_deals([self.deal('base', 'd1', 100, 10)]) book.load() - self.assertEqual(book.items['600000.SH']['base_qty'], 100) + self.assertEqual(book.state['600000.SH']['base_qty'], 100) self.assertEqual(book.deals_sys_ids, {'d1'}) 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) - state.reconcile([], [self.deal('sell', 'd1', 100, 12)]) - item = state.items['600000.SH'] - self.assertEqual((item.base_qty, item.sell_qty, item.phase), (100, 100, SOLD)) - state.reconcile([], [self.deal('buy', 'd2', 100, 11)]) - self.assertEqual(TState(self.path).items['600000.SH'].phase, DONE) - def test_failed_insert_rolls_back_memory_and_database(self): - state = TState(self.path) - with closing(sqlite3.connect(self.path)) as db: - db.execute("""CREATE TRIGGER fail_insert BEFORE INSERT ON deals - BEGIN SELECT RAISE(ABORT, 'test failure'); END""") - fill = self.deal('base', 'd1', 100, 10) - with self.assertRaises(sqlite3.IntegrityError): - state.reconcile([], [fill]) - self.assertFalse(state.items) - self.assertFalse(state.deals) - self.assertFalse(TState(self.path).items) - with closing(sqlite3.connect(self.path)) as db: - db.execute('DROP TRIGGER fail_insert') - state.reconcile([], [fill]) - self.assertEqual(TState(self.path).items['600000.SH'].base_qty, 100) - def test_schema_and_unique_execution(self): - state = TState(self.path) - 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, {'state', 'deals', 'sqlite_sequence'}) - columns = {row[1] for row in db.execute('PRAGMA table_info(deals)')} - 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_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']), - ('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])) - with self.assertRaises(sqlite3.IntegrityError): - state.save() - self.assertEqual(len(state.deals), 1) - state.items['600000.SH'].base_cost = float('inf') - with self.assertRaises(ValueError): - state.save() - self.assertEqual(state.items['600000.SH'].base_cost, 10) def test_position_columns_defaults_indexes_and_stable_id(self): store = State(self.path) @@ -176,7 +93,7 @@ class OrderBookTests(unittest.TestCase): position = PositionItem(stock_code='600000.SH', volume=100, open_price=10, stock_name='stock', can_use_volume=100, float_profit=-2.5) store.sync_state([position]) - saved = store.items[position.stock_code] + saved = store.state[position.stock_code] first_id = saved['id'] self.assertEqual(saved['base_qty'], 100) self.assertEqual(saved['base_price'], 10) @@ -186,20 +103,20 @@ class OrderBookTests(unittest.TestCase): position.volume = 200 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) + self.assertEqual(store.state[position.stock_code]['id'], first_id) + self.assertEqual(store.state[position.stock_code], saved) + self.assertEqual(State(self.path).state[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) + self.assertEqual(store.state[position.stock_code], saved) + self.assertEqual(store.state['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, {}) + self.assertEqual(store.state, {}) + self.assertEqual(State(self.path).state, {}) store.sync_state([PositionItem(stock_code='600001.SH', volume=100)]) - self.assertGreater(store.items['600001.SH']['id'], first_id) + self.assertGreater(store.state['600001.SH']['id'], first_id) store.sync_state([]) - self.assertEqual(store.items, {}) + self.assertEqual(store.state, {}) self.assertEqual(store.deals, saved_deals) def test_state_fields_survive_restart_and_sync(self): @@ -211,37 +128,23 @@ class OrderBookTests(unittest.TestCase): 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']] + with closing(book._connect()) as db, db: + db.execute( + f"INSERT INTO state ({', '.join(row)}) VALUES ({', '.join(':' + key for key in row)})", + row, + ) + book.load() + saved = book.state[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 + self.assertEqual(book.state[row['stock_code']], saved) with self.assertRaises(sqlite3.IntegrityError): - book.save({row['stock_code']: row}) - self.assertEqual(State(self.path).items[row['stock_code']], saved) + with closing(book._connect()) as db, db: + db.execute('UPDATE state SET added_qty = -1') + self.assertEqual(State(self.path).state[row['stock_code']], saved) - def test_base_split_fills_and_snapshot_do_not_double_count(self): - state = TState(self.path) - first = self.deal('base', 'd1', 40, 10) - state.reconcile([PositionItem(stock_code='600000.SH', volume=40, open_price=10)], [first]) - second = self.deal('base', 'd2', 60, 12) - state.reconcile([PositionItem(stock_code='600000.SH', volume=100, open_price=11.2)], [first, second]) - self.assertEqual(state.items['600000.SH'].base_qty, 100) - self.assertAlmostEqual(state.items['600000.SH'].base_cost, 11.2) - self.assertEqual(len(state.deals), 2) - def test_date_normalization_and_unrelated_strategy(self): - state = TState(self.path) - first = self.deal('base', 'd1', 100, 10, '20260901') - other = self.deal('base', 'd2', 100, 10) - other.remark = 'trend-base-order' - state.reconcile([], [first, other]) - first.trade_date = '2026-09-01' - state.reconcile([], [first]) - self.assertEqual(len(state.deals), 1) - self.assertEqual(state.deals[0]['trade_date'], '2026-09-01') if __name__ == '__main__': diff --git a/py-client/tests/test_state_archiving.py b/py-client/tests/test_state_archiving.py index 65a7a38..2dcaab7 100644 --- a/py-client/tests/test_state_archiving.py +++ b/py-client/tests/test_state_archiving.py @@ -22,6 +22,11 @@ class ArchivingTests(unittest.TestCase): (code, order, order, flag, amount / qty, qty, amount, '2026-09-08', time), ) + def test_schema_only_has_state_and_deals(self): + with closing(self.book._connect()) as db: + tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} + self.assertEqual(tables, {'state', 'deals', 'sqlite_sequence'}) + 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']) @@ -45,7 +50,7 @@ class ArchivingTests(unittest.TestCase): 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') + self.assertEqual(row['added_order_local_id'], 'late') def test_sell_added_then_clear_base(self): self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) @@ -80,11 +85,7 @@ class ArchivingTests(unittest.TestCase): 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) + def test_excess_sell_rolls_back(self): self.insert_deal('buy', 50, 500, '10:00:00') self.insert_deal('sell', 100, 1200, '10:01:00', flag=49) errors = self.book.archiving() @@ -93,6 +94,17 @@ class ArchivingTests(unittest.TestCase): self.assertEqual(restarted.state, {}) self.assertTrue(all(deal['is_arch'] == 0 for deal in restarted.deals.values())) + def test_stock_buy_and_sell_flags(self): + self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) + self.insert_deal('buy', 100, 1000, '10:00:00', flag=23) + self.insert_deal('sell', 50, 600, '10:01:00', flag=24) + self.assertEqual(self.book.archiving(), {}) + row = self.book.state['600000.SH'] + self.assertEqual((row['base_qty'], row['added_qty']), (100, 50)) + self.assertEqual(self.book.deals['buy']['offset_flag'], 23) + self.assertEqual(self.book.deals['sell']['offset_flag'], 24) + self.assertTrue(all(deal['is_arch'] == 1 for deal in self.book.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') @@ -115,7 +127,7 @@ class ArchivingTests(unittest.TestCase): 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): + def test_equal_quantity_buy_is_added_and_preserves_status(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', @@ -124,7 +136,7 @@ class ArchivingTests(unittest.TestCase): )]) self.assertEqual(self.book.archiving(), {}) row = self.book.state['600000.SH'] - self.assertEqual((row['base_qty'], row['added_qty']), (100, 0)) + self.assertEqual((row['base_qty'], row['added_qty']), (100, 100)) self.assertEqual(self.book.deals['first']['is_arch'], 1) restarted = State(self.book.path) self.assertEqual(restarted.archiving(), {}) @@ -134,19 +146,20 @@ class ArchivingTests(unittest.TestCase): 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')) + self.assertEqual((row['base_qty'], row['added_qty'], row['status']), (100, 200, 'CUSTOM')) - def test_old_archived_history_without_baseline_is_not_reapplied(self): + def test_archived_history_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(errors, {}) + self.assertEqual(self.book.state['600000.SH']['added_qty'], 50) self.assertEqual(self.book.deals['old']['is_arch'], 1) - self.assertEqual(self.book.deals['new']['is_arch'], 0) + self.assertEqual(self.book.deals['new']['is_arch'], 1) - def test_late_buy_replays_after_liquidation_and_restart(self): + def test_late_buy_is_incremental_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() @@ -156,11 +169,12 @@ class ArchivingTests(unittest.TestCase): self.assertEqual(self.book.archiving(), {}) row = self.book.state['600000.SH'] self.assertEqual(row['added_qty'], 100) - self.assertEqual(row['added_price'], 15) + self.assertEqual(row['added_price'], 20) - def test_snapshot_deletion_before_sell_archiving(self): + def test_archive_sell_before_syncing_empty_positions(self): self.book.sync_state([PositionItem(stock_code='600000.SH', volume=100, open_price=8)]) self.insert_deal('sell', 100, 1000, '10:01:00', flag=49) + self.assertEqual(self.book.archiving(), {}) self.book.sync_state([]) self.book = State(self.book.path) self.assertEqual(self.book.archiving(), {}) diff --git a/py-client/tests/test_zt_state.py b/py-client/tests/test_zt_state.py new file mode 100644 index 0000000..c0411df --- /dev/null +++ b/py-client/tests/test_zt_state.py @@ -0,0 +1,54 @@ +import tempfile +import unittest +from pathlib import Path + +from libs.state import State +from sdk import DealItem, PositionItem +from strategy.zt.boot import sync_account_state + + +class ZTStateTests(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.state = State(Path(tmp.name) / 'zt_test_state.db') + + def position(self, qty): + return PositionItem(stock_code='600000.SH', volume=qty, open_price=10) + + def deal(self, order, qty, flag=23, strategy='zt'): + return DealItem( + stock_code='600000.SH', order_sys_id=order, + remark=f'{strategy}-buy-{order}|{strategy}', offset_flag=flag, + volume=qty, price=10, trade_amount=qty * 10, + trade_date='20260909', trade_time='100000', + ) + + def test_initial_snapshot_and_incremental_deals_after_restart(self): + historical = self.deal('old', 100) + unrelated = self.deal('trend', 100, strategy='trend') + sync_account_state(self.state, [self.position(100)], [historical, unrelated], initialize=True) + self.assertEqual(set(self.state.deals), {'old'}) + self.assertEqual(self.state.state['600000.SH']['base_qty'], 100) + self.assertEqual(self.state.state['600000.SH']['added_qty'], 0) + self.state = State(self.state.path) + bought = self.deal('new', 100, flag=48) + for _ in range(2): + sync_account_state(self.state, [self.position(200)], [historical, bought, unrelated]) + row = self.state.state['600000.SH'] + self.assertEqual((row['base_qty'], row['added_qty']), (100, 100)) + sold = self.deal('sell', 200, flag=24) + sync_account_state(self.state, [], [historical, bought, sold]) + self.assertEqual(self.state.state, {}) + self.assertEqual(self.state.deals['sell']['is_arch'], 1) + + def test_archive_failure_preserves_holdings_for_retry(self): + sync_account_state(self.state, [self.position(100)], [], initialize=True) + with self.assertRaisesRegex(ValueError, 'ZT'): + sync_account_state(self.state, [], [self.deal('sell', 200, flag=49)]) + self.assertEqual(self.state.state['600000.SH']['base_qty'], 100) + self.assertEqual(self.state.deals['sell']['is_arch'], 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/py-client/tests/test_zt_trading.py b/py-client/tests/test_zt_trading.py new file mode 100644 index 0000000..2784510 --- /dev/null +++ b/py-client/tests/test_zt_trading.py @@ -0,0 +1,136 @@ +import tempfile +import unittest +from datetime import datetime +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from config import AccountConfig +from libs.grid_take_profit import GridState +from libs.state import State +from sdk import Assets, DealItem, PositionItem, Tick +from strategy.zt import boot +from strategy.zt.open import open_signal +from strategy.zt.positions import manage_positions, t_rounds + + +class ZTTradingTests(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.store = State(Path(tmp.name) / 'state.db') + self.code = '600000.SH' + self.cfg = AccountConfig(account_id='test', buy_value=2000, zt_sell_ratio=0.5) + self.run = SimpleNamespace(account_cfg=self.cfg, orders=Mock(), client=Mock(), + profit_tracker=Mock(), add_watch=Mock(), open_watch=Mock()) + self.run.orders.busy.return_value = False + self.run.orders.new_order_id.side_effect = lambda kind: f'zt-{kind}-order' + self.run.profit_tracker.observe.return_value.state = GridState.RETREAT + self.run.add_watch.triggered.return_value = True + self.run.open_watch.triggered.return_value = True + self.position = PositionItem(stock_code=self.code, volume=200, can_use_volume=200, open_price=10) + boot.sync_account_state(self.store, [self.position], [], initialize=True) + + def fill(self, kind, order, qty, price=10, date='2026-09-09'): + return DealItem(stock_code=self.code, order_sys_id=order, remark=f'zt-{kind}-{order}|zt', + offset_flag=24 if kind == 't-sell' else 23, + volume=qty, price=price, trade_amount=qty * price, + trade_date=date, trade_time='100000') + + def manage(self, price=11, available=10000, positions=None, force=False, today='2026-09-09'): + return manage_positions(self.run, self.store, {self.code: Tick(last_price=price)}, + [self.position] if positions is None else positions, + t_rounds(self.store), available, today, force) + + def test_sell_only_available_shares_and_no_loss_sell(self): + self.position.can_use_volume = 0 + self.manage() + self.run.orders.place.assert_not_called() + self.position.can_use_volume = 100 + self.manage(price=9) + self.run.orders.place.assert_not_called() + self.manage() + request = self.run.orders.place.call_args.args[1] + self.assertEqual((request.op, request.volume), (24, 100)) + + def test_full_sale_restart_and_force_buyback_without_price_or_market_gate(self): + sell = self.fill('t-sell', 's1', 200, price=11) + boot.sync_account_state(self.store, [], [sell]) + self.store = State(self.store.path) + self.cfg.zt_max_price = 10 + self.run.add_watch.triggered.return_value = False + remaining = self.manage(price=12, positions=[], force=True) + request = self.run.orders.place.call_args.args[1] + self.assertEqual((request.op, request.volume), (23, 200)) + self.assertAlmostEqual(remaining, 10000 - 12 * 200 * 1.01) + + def test_partial_fills_once_and_completed_round_blocks_same_day_sale(self): + deals = [self.fill('t-sell', 's1', 40, 11), self.fill('t-sell', 's2', 60, 12)] + self.position.volume = 100 + boot.sync_account_state(self.store, [self.position], deals + deals) + item = t_rounds(self.store)[self.code] + self.assertEqual(item['sold'], 100) + self.assertEqual(item['amount'], 1160) + self.manage(price=10) + self.assertEqual(self.run.orders.place.call_args.args[1].volume, 100) + deals.append(self.fill('t-buy', 'b1', 100)) + self.position.volume = 200 + boot.sync_account_state(self.store, [self.position], deals) + self.run.orders.place.reset_mock() + self.manage(price=11) + self.run.orders.place.assert_not_called() + self.manage(price=11, today='2026-09-10') + self.assertEqual(self.run.orders.place.call_args.args[1].op, 24) + + def test_cross_day_debt_and_insufficient_cash(self): + boot.sync_account_state(self.store, [], [self.fill('t-sell', 's1', 200, date='2026-09-08')]) + self.manage(positions=[], available=100, force=True) + self.run.orders.place.assert_not_called() + self.manage(positions=[], force=True) + self.assertEqual(self.run.orders.place.call_args.args[1].volume, 200) + + def test_delayed_snapshot_does_not_delete_or_recreate_holdings(self): + boot.sync_account_state(self.store, [], []) + self.assertEqual(self.store.state[self.code]['base_qty'], 200) + sell = self.fill('t-sell', 's1', 200) + boot.sync_account_state(self.store, [self.position], [sell]) + self.assertNotIn(self.code, self.store.state) + self.manage() + self.run.orders.place.assert_not_called() + + def test_base_fills_stay_in_base_bucket(self): + self.store.sync_state([]) + deals = [self.fill('base', 'b1', 100), self.fill('base', 'b2', 100, 12)] + boot.sync_account_state(self.store, [self.position], deals) + row = self.store.state[self.code] + self.assertEqual((row['base_qty'], row['base_price'], row['added_qty']), (200, 11, 0)) + + def test_run_once_queries_sold_out_code_and_never_opens_with_debt(self): + sell = self.fill('t-sell', 's1', 200, 11) + self.run.client.deals.return_value = [sell] + self.run.client.portfolio.return_value = SimpleNamespace(assets=Assets(10000, 10000), positions={}, orders=[]) + self.run.client.full_tick.return_value = {self.code: Tick(last_price=12)} + with patch.object(boot, 'datetime') as clock, patch.object(boot, 'collector_push'), \ + patch.object(boot, 'open_signal') as opened, patch.object(boot, 'market_allow_open') as market: + clock.now.return_value = datetime(2026, 9, 9, 14, 50) + boot.RunOnce(self.run, self.store, []) + self.run.client.full_tick.assert_called_once_with([self.code]) + opened.assert_not_called() + market.assert_not_called() + self.assertEqual(self.run.orders.place.call_args.args[1].op, 23) + + def test_open_budget_includes_buffer_and_star_minimum(self): + with patch('strategy.zt.open.datetime') as clock: + clock.now.return_value = datetime(2026, 9, 9, 10) + remaining = open_signal(self.run, {self.code: Tick(last_price=10)}, + [SimpleNamespace(code=self.code)], 2000) + self.assertEqual(self.run.orders.place.call_args.args[1].volume, 100) + self.assertEqual(remaining, 990) + self.run.orders.place.reset_mock() + open_signal(self.run, {'688001.SH': Tick(last_price=10)}, + [SimpleNamespace(code='688001.SH')], 2000) + self.run.orders.place.assert_not_called() + + +if __name__ == '__main__': + unittest.main()