From c7d39938a22c5b558441c01003a69fb765a5011e Mon Sep 17 00:00:00 2001 From: yanweidong Date: Thu, 17 Sep 2026 00:56:05 +0800 Subject: [PATCH] add etf --- py-client/README.md | 5 + py-client/config/__init__.py | 7 + py-client/etc/etf.yaml | 23 ++ py-client/libs/order.py | 4 + py-client/main.py | 2 + py-client/strategy/etf/README.md | 131 +++++++++++ py-client/strategy/etf/__init__.py | 1 + py-client/strategy/etf/boot.py | 67 ++++++ py-client/strategy/etf/config.py | 59 +++++ py-client/strategy/etf/data.py | 55 +++++ py-client/strategy/etf/engine.py | 202 ++++++++++++++++ py-client/strategy/etf/indicators.py | 55 +++++ py-client/strategy/etf/state.py | 65 ++++++ py-client/tests/test_etf.py | 329 +++++++++++++++++++++++++++ 14 files changed, 1005 insertions(+) create mode 100644 py-client/etc/etf.yaml create mode 100644 py-client/strategy/etf/README.md create mode 100644 py-client/strategy/etf/__init__.py create mode 100644 py-client/strategy/etf/boot.py create mode 100644 py-client/strategy/etf/config.py create mode 100644 py-client/strategy/etf/data.py create mode 100644 py-client/strategy/etf/engine.py create mode 100644 py-client/strategy/etf/indicators.py create mode 100644 py-client/strategy/etf/state.py create mode 100644 py-client/tests/test_etf.py diff --git a/py-client/README.md b/py-client/README.md index 3774195..219871a 100644 --- a/py-client/README.md +++ b/py-client/README.md @@ -62,6 +62,11 @@ ZT 轮次状态机（正T/反T）、Trend 采集任务与 Python 3.14 回归。 修改前 18 项测试中 12 项失败，原因是模型仅有 `get_local_order_id` 属性，调用处却使用缺失的 `local_order_id`，存储层还将属性当方法调用。 本次增加同一属性的兼容别名，并统一存储层属性访问，保留原属性名和 API 数据字段；这些是使既有撤单、成交对账测试恢复的接口修复。 +## ETF 自适应网格策略 + +入口为 `strategy: etf`，标的和参数见 [`etc/etf.yaml`](etc/etf.yaml)， +完整规则和启用步骤见 [`strategy/etf/README.md`](strategy/etf/README.md)。 + ## ZT 做 T 策略（2026-09 重构） ZT 已从"本地 SQLite 重算持仓 + base/added 分桶归档"改为**正T/反T 轮次状态机**： diff --git a/py-client/config/__init__.py b/py-client/config/__init__.py index a0770f7..1497c7e 100644 --- a/py-client/config/__init__.py +++ b/py-client/config/__init__.py @@ -60,6 +60,8 @@ class AccountConfig: # 当前账户启用的策略名称，例如 trend。 strategy: str = "" + # 为空时读取 py-client/etc/etf.yaml；非空路径相对于账户配置目录。 + etf_config_path: str = "" # load() 成功后保存已加载的配置，供策略模块直接读取。 @@ -132,6 +134,11 @@ def load( Path(global_config.qmt_data_dir).mkdir(parents=True, exist_ok=True) account_config = AccountConfig(**_account_values(root / account_file)) + if account_config.etf_config_path: + etf_path = Path(account_config.etf_config_path) + account_config.etf_config_path = str(etf_path if etf_path.is_absolute() else root / etf_path) + elif account_config.strategy.strip().lower() == 'etf': + account_config.etf_config_path = str(root / 'etf.yaml') if account_config.buy_value <= 0 or account_config.grid_step_pct <= 0: raise ValueError("buy_value、grid_step_pct 必须大于 0") if type(account_config.zt_open_hands) is not int or account_config.zt_open_hands < 0: diff --git a/py-client/etc/etf.yaml b/py-client/etc/etf.yaml new file mode 100644 index 0000000..1ba6663 --- /dev/null +++ b/py-client/etc/etf.yaml @@ -0,0 +1,23 @@ +# 标的是示例白名单；仅在账户配置 strategy: etf 时启用。 +codes: + - "510300.SH" + - "159915.SZ" +# 固定买入手数，每手 100 份；每只 ETF 总持仓硬上限 10 手。 +buy_hands: 1 +max_hands: 10 +# 60 日均线作为中轴；格距 = max(ATR×倍数, MA60×最小格距百分比, 0.001)。 +atr_period: 14 +atr_multiplier: 1.0 +boll_period: 20 +boll_std: 2.0 +min_grid_pct: 0.5 +# 复用 DipWatch：触及低位后，从观察低点反弹 0.61% 才买入。 +rebound_pct: 0.61 +watch_seconds: 600 +# 达到高位和最低利润要求后启动网格回撤止盈。 +min_profit_pct: 0.5 +# 用于资金预留及止盈费用门槛，按实际券商佣金调整。 +commission_rate: 0.0003 +min_commission: 5.0 +# 超过此秒数或缺少时间戳的行情不交易。 +max_tick_age_seconds: 90 diff --git a/py-client/libs/order.py b/py-client/libs/order.py index 3bc6517..f43ab1a 100644 --- a/py-client/libs/order.py +++ b/py-client/libs/order.py @@ -27,6 +27,8 @@ class PlaceOrderRequest: order_id: str strategy_name: str kind: str = "" + # ETF 使用 0.001 元精度的限价；未指定时保留原策略的最新价委托。 + price: float | None = None class OrderBook: @@ -129,12 +131,14 @@ class OrderBook: self.busy_cache.set(key, True, timeout=self.lock_timeout_sec) try: + price_args = {} if request.price is None else {"pr_type": 11, "price": request.price} result = client.passorder( op_type=request.op, stock_code=request.code, volume=request.volume, strategy_name=request.strategy_name, order_id=request.order_id, + **price_args, ) except APIError as exc: logging.exception( diff --git a/py-client/main.py b/py-client/main.py index 7c85ad5..7bc7711 100644 --- a/py-client/main.py +++ b/py-client/main.py @@ -37,6 +37,7 @@ 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.etf.boot import StartETF from strategy.ipo import AutoBuyIpo @dataclass(slots=True) @@ -48,6 +49,7 @@ class StrategyDefinition: STRATEGIES = { "trend": StrategyDefinition("Trend", StartTrend), "zt": StrategyDefinition("ZT", StartZT), + "etf": StrategyDefinition("ETF", StartETF), } def require_windows() -> bool: diff --git a/py-client/strategy/etf/README.md b/py-client/strategy/etf/README.md new file mode 100644 index 0000000..aa8d554 --- /dev/null +++ b/py-client/strategy/etf/README.md @@ -0,0 +1,131 @@ +# A 股 ETF 自适应网格策略 + +策略目录为 `py-client/strategy/etf`，入口名称 `etf`，独立配置为 +`py-client/etc/etf.yaml`。每只 ETF 总持仓上限 10 手（1000 份），默认每次买入 1 手。 +这是常见均值回归与波动率网格方法的工程组合，尚未完成历史收益回测。 + +## 方法分析 + +| 方法 | 优点 | 局限 | 本策略选择 | +| --- | --- | --- | --- | +| 固定价差网格 | 简单直观 | 不适应不同价格与波动率 | ATR 动态格距，设置百分比下限 | +| 均线回归 | 提供相对高低位置 | 单边下跌中均线会滞后 | MA60 作中轴，设置仓位上限 | +| BOLL 低吸 | 用价格分布寻找相对低位 | 触及下轨不代表跌势结束 | 下轨仅启动观察，反弹后再买 | +| 回撤止盈 | 上涨时跟随峰值 | 不能保证最高价退出 | 复用现有 GridTrailingTracker | + +ATR 反映波动幅度，不判断方向；BOLL 为均线加减标准差倍数。定义参考 +[Fidelity ATR](https://www.fidelity.com/learning-center/trading-investing/technical-analysis/technical-indicator-guide/atr) +及 [Fidelity BOLL](https://www.fidelity.com/learning-center/trading-investing/technical-analysis/technical-indicator-guide/bollinger-bands)。 +默认参数是可调整的起点，不代表已经优化或保证收益。 + +## 指标与网格 + +历史日线直接读取 `http://go.apinb.com/a/get_daily?code=<证券代码>`，使用响应 +`details` 中的 `ts_code / trade_date / open / high / low / close`。校验 `code=0`、 +证券归属、日期和 OHLC 后，按日期排序，排除当天及未来数据，再取最近 120 根。 +每个标的每天计算一次并缓存，至少需要 60 根；ATR 周期为 60 时至少需要 61 根。 +接口样本未声明复权口径，策略使用接口原始价格，不自行假定或执行前复权。 + +- `M = 最近60根收盘价平均值`，固定 MA60。 +- `TR = max(最高-最低, abs(最高-前收盘), abs(最低-前收盘))`。 +- ATR 默认 14 日：前 14 个 TR 平均作初值，后续按 + `ATR = (前ATR × 13 + 当日TR) / 14` 进行 Wilder 平滑。 +- BOLL 默认 20 日、2 倍总体标准差：`中轨=MA20`，`上下轨=MA20 ± 2σ`。 +- 格距 `G = max(ATR × atr_multiplier, M × min_grid_pct / 100, 0.001)`， + 向上取整至 0.001 元。均线作中轴，ATR 决定每格宽度。 + +重复日期、非正数或非有限价格、价格关系异常、日线不足均不交易。 +最后日线超过 15 个自然日也不交易；该检查只排除明显过期，不能替代交易所日历。 +接口数据应更新至上一交易日。获取失败的标的每 5 分钟重试。 + +## 买入规则 + +1. 首仓在 `现价 <= min(BOLL下轨, M-G)` 时开始观察，不直接买入。 +2. 复用 `DipWatch` 防接飞刀：下跌刷新低点，从低点反弹默认 0.61% 后触发； + 观察默认 600 秒过期重置。反弹可站回下轨上方，但不能超过 MA60。 +3. 加仓还须满足 `现价 <= 上次实际买入成交均价-G`，防止同价位连续补满。 + 初次接管已有仓位时，券商成本作为初始加仓基准。 +4. 每次买入 `buy_hands × 100` 份；若买后超过 `max_hands × 100`，整笔跳过， + 不临时缩量。配置强制 `1 <= buy_hands <= max_hands <= 10`。 +5. 买入预算为券商可用资金减账户现金安全线，再减所有本地待确认买单。 + 限价金额加预估佣金占用预算，多标的串行扣减。 + +示例：MA60=4.00、G=0.05、下轨=3.90，价格进入 3.90 以下才观察；低点 3.88 +反弹至 3.904 时超过 0.61%，可以提交固定手数。如果实际均价为 3.904， +下一笔最高买价为 3.854，同时仍须满足低位触发和反弹确认。 + +## 止盈规则 + +1. `现价 >= max(BOLL上轨, M+G, 成本+G)`，且满足 `min_profit_pct`、 + 预估价差收益大于买卖两侧佣金，才启动高位跟踪。 +2. 启动时冻结格距。盈利格编号为 `floor((现价-成本)/冻结格距)`，复用 + `GridTrailingTracker` 记录最高格；进入更高格更新峰值，跌回较低格时止盈。 + 这是跌破峰值格边界，不是从最高价回撤完整一个 ATR。 +3. 启动后，即使回落到 BOLL 上轨或启动价以下，也继续判断回撤；卖出时仍须满足 + 最低利润和费用门槛。止盈启动后暂停补仓，不同时发买卖单。 +4. **固定手数用于买入；止盈卖出当前全部可用整手份额**，数量为 + `min(持仓, 券商可卖数量)` 向下取整到 100 份。零股暂不处理。 +5. T+1 当天不可卖时继续记录峰值，翌日按券商 `can_use_volume` 判断。 + 清仓、实际成交改变仓位、外部数量或成本变化后重建基准；部分卖出后剩余仓位 + 重新等待高位启动，不把旧峰值带入新仓。 + +交易单位、0.001 元报价精度及股票 ETF 的 T+1 参考 +[上交所 ETF 常见问题](https://www.sse.com.cn/assortment/fund/etf/question/)。 +本策略没有自动止损，单边下跌可能满仓后长期持有；10 手上限只限制数量。 +佣金参数按券商实际情况调整。费用门槛是估算，不逐笔归集历史买入最低佣金； +若券商持仓成本已含费用，该估算会偏保守。 + +## 委托、持仓与持久化 + +- 管理 `codes` 白名单内的已有持仓；`excluded_codes` 优先排除。配置外证券不买卖。 +- 以 `ETF-BUY-*` / `ETF-SELL-*` 为本地编号，标签 `etf`，当前价按 0.001 元 + 精度提交限价。复用 OrderBook，只自动撤销超时 120 秒的 ETF 前缀委托。 +- 同标的全账户买卖在途、未知委托状态或在途份额都会阻止新单。 + 行情缺少 `timetag`、不是当天或超过默认 90 秒，也不交易。 +- 下单前原子保存意图，HTTP 成功、超时或失败均不会自动解除锁。 + 必须收到终态（53/54/56/57），并且券商持仓与累计成交量相符，才能继续。 + 部分成交按实际数量核对，买入必须取得实际成交均价才推进下一格。 +- 状态位于 `{qmt_data_dir}/etf/{账户SHA256}/state.json`，保存实际买入基准、 + 止盈峰值、冻结格距和未确认委托；重启恢复，损坏不静默覆盖。 +- 同一 ETF 不适合同时由人工或其他策略频繁交易。在途期间外部改变持仓，会暂停核对。 + 若跨日后柜台不再返回未确认订单，则持续暂停该标的，需要核对历史订单与持仓后 + 人工处理状态，不按超时自动重发。有未确认委托的标的不能直接从配置移除。 + +## 启用 + +1. 修改 `etc/etf.yaml` 的标的及参数。示例仅示范格式，请选择实际交易的 A 股股票 ETF； + 代码形态检查不验证基金投资范围。 +2. 主机对应的账户 YAML 设置： + + ```yaml + strategy: etf + etf_config_path: etf.yaml + enable_auto_ipo: false + ``` + + 路径相对账户配置目录，也支持绝对路径；不填默认 `etc/etf.yaml`。 + `account_id`、`min_cash_ratio` 继续生效。公共校验仍要求 `buy_value`、 + `grid_step_pct` 为正，但 ETF 不用它们计算数量或格距。 + 关闭 IPO 是此处示例选择；ETF 决策不依赖 IPO 或远端股票信号。 +3. 无需修改 QMT 服务端或 `sdk/`。历史接口适配完全位于 `strategy/etf/data.py`； + 实时行情、持仓和交易继续使用已有 QMT SDK。 +4. 确认外部接口提供配置 ETF 的足量、最新日线。HTTP 错误、业务失败、空列表、 + 证券代码不一致或数据异常都会跳过该标的，不改用其他证券数据。 + 2026-09-17 联通验证中，股票样本 `600584.SH` 成功返回 200 条，示例 ETF + `510300.SH` 返回 404；需要数据服务覆盖实际配置的 ETF 后才能正常运行。 +5. 按现有方式运行 `python main.py`。每 30 秒执行，午休暂停，15:00 退出。 + +本次新增不会自动切换已有实盘账户，也没有进行实盘委托。 + +## 离线验证 + +在 `py-client` 中运行： + +```powershell +python -m unittest discover -s tests -p test_etf.py -v +python -m unittest discover -s tests -v +``` + +覆盖指标、未收盘日线排除、反弹确认、固定手数、限仓、资金共享、部分成交、拒单、 +快照延迟、T+1、重启防重、峰值恢复、状态损坏、外部日线接口与异常响应。 +离线行为验证不等同于历史收益回测或实盘联调。 diff --git a/py-client/strategy/etf/__init__.py b/py-client/strategy/etf/__init__.py new file mode 100644 index 0000000..d2a32fb --- /dev/null +++ b/py-client/strategy/etf/__init__.py @@ -0,0 +1 @@ +"""A 股场内 ETF：均线中轴、ATR 网格与 BOLL 低吸策略。""" diff --git a/py-client/strategy/etf/boot.py b/py-client/strategy/etf/boot.py new file mode 100644 index 0000000..f7ca618 --- /dev/null +++ b/py-client/strategy/etf/boot.py @@ -0,0 +1,67 @@ +"""ETF 策略入口：每 30 秒运行，日线指标当天缓存，失败标的单独重试。""" + +from datetime import datetime, timedelta +import hashlib +import logging as log +from pathlib import Path +import time +import httpx + +import config +from libs.calc import trading_time +from libs.snapshot import cache_portfolio +from sdk import Client + +from .config import load +from .data import daily_bars +from .engine import Engine +from .indicators import calculate +from .state import Store + + +def StartETF() -> None: + cfg = load(config.account_config.etf_config_path or None) + account = str(config.account_config.account_id).strip() + if not account: + raise ValueError('ETF 策略缺少账户编号') + key = hashlib.sha256(account.encode('utf-8')).hexdigest() + store = Store(Path(config.global_config.qmt_data_dir) / 'etf' / key / 'state.json', account) + # 独立 HTTP 连接池读取外部日线，不向外部接口发送 QMT 认证信息。 + with Client(config.global_config.qmt_base_url, config.global_config.qmt_token, config.HTTP_TIMEOUT) as client, \ + httpx.Client(timeout=config.HTTP_TIMEOUT) as history_client: + engine = Engine(client, cfg, store, config.account_config.min_cash_ratio, + config.account_config.excluded_codes) + log.info('[ETF启动] 标的=%s 每次=%d手 每只上限=%d手 状态=%s', + cfg.codes, cfg.buy_hands, cfg.max_hands, store.path) + log.info('[ETF启动] 管理配置白名单内已有持仓，卖出以券商可用份额为限') + indicators, retry_at = {}, {} + cached_day = None + while True: + now = datetime.now() + if now.hour >= 15: + log.info('[ETF结束] 已到 15:00') + return + if trading_time(now): + try: + if cached_day != now.date(): + indicators, retry_at, cached_day = {}, {}, now.date() + for code in cfg.codes: + if code in indicators or now < retry_at.get(code, datetime.min): + continue + try: + rows = daily_bars(history_client, code, now.date()) + indicators[code] = calculate(rows, now.date(), cfg) + log.info('[ETF指标] %s %s', code, indicators[code]) + except Exception: + retry_at[code] = now + timedelta(minutes=5) + log.exception('[ETF日线] %s 获取或计算失败，5分钟后重试', code) + portfolio = client.portfolio() + ticks = client.full_tick(list(cfg.codes)) + engine.run(portfolio, ticks, indicators, datetime.now()) + try: + cache_portfolio(account, portfolio.assets, list(portfolio.positions.values()), client.deals()) + except Exception: + log.exception('[ETF采集] 成交快照读取失败') + except Exception: + log.exception('[ETF异常] 本轮失败，下一轮继续') + time.sleep(30 - datetime.now().second % 30) diff --git a/py-client/strategy/etf/config.py b/py-client/strategy/etf/config.py new file mode 100644 index 0000000..1b88183 --- /dev/null +++ b/py-client/strategy/etf/config.py @@ -0,0 +1,59 @@ +"""独立读取 ETF 参数，不修改现有账户策略的默认行为。""" + +from dataclasses import dataclass, fields +from pathlib import Path +import math +import re +import yaml + + +@dataclass(frozen=True) +class ETFConfig: + codes: tuple[str, ...] = () + buy_hands: int = 1 + max_hands: int = 10 + atr_period: int = 14 + atr_multiplier: float = 1.0 + boll_period: int = 20 + boll_std: float = 2.0 + min_grid_pct: float = 0.5 + rebound_pct: float = 0.61 + watch_seconds: int = 600 + min_profit_pct: float = 0.5 + commission_rate: float = 0.0003 + min_commission: float = 5.0 + max_tick_age_seconds: int = 90 + + def __post_init__(self): + # 限定沪深场内 ETF 代码形态；具体跟踪 A 股的标的由配置白名单决定。 + if not isinstance(self.codes, (list, tuple)) or not self.codes: + raise ValueError('etf.yaml 的 codes 必须是非空 ETF 代码列表') + if any(not isinstance(c, str) or not re.fullmatch(r'(?:5[0-9]{5}\.SH|1[58][0-9]{4}\.SZ)', c) + for c in self.codes) or len(set(self.codes)) != len(self.codes): + raise ValueError('ETF 代码必须唯一，使用完整沪深场内代码，如 510300.SH、159915.SZ') + object.__setattr__(self, 'codes', tuple(self.codes)) + for name in ('buy_hands', 'max_hands', 'atr_period', 'boll_period', 'watch_seconds', 'max_tick_age_seconds'): + if type(getattr(self, name)) is not int or getattr(self, name) <= 0: + raise ValueError(f'{name} 必须为正整数') + if not self.buy_hands <= self.max_hands <= 10: + raise ValueError('必须满足 buy_hands <= max_hands <= 10，每手 100 份') + if not 2 <= self.atr_period <= 60 or not 2 <= self.boll_period <= 60: + raise ValueError('ATR、BOLL 周期必须在 2 到 60 日之间') + for name in ('atr_multiplier', 'boll_std', 'min_grid_pct', 'rebound_pct', 'min_profit_pct', + 'commission_rate', 'min_commission'): + value = getattr(self, name) + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0: + raise ValueError(f'{name} 必须是有限非负数') + if name not in ('commission_rate', 'min_commission') and value == 0: + raise ValueError(f'{name} 必须大于零') + + +def load(path: str | Path | None = None) -> ETFConfig: + path = Path(path) if path else Path(__file__).resolve().parents[2] / 'etc' / 'etf.yaml' + try: + raw = yaml.safe_load(path.read_text(encoding='utf-8')) + except (OSError, yaml.YAMLError) as exc: + raise ValueError(f'ETF 配置读取失败：{path}') from exc + if not isinstance(raw, dict) or set(raw) - {f.name for f in fields(ETFConfig)}: + raise ValueError('ETF 配置必须为对象，且不能包含未知参数') + return ETFConfig(**raw) diff --git a/py-client/strategy/etf/data.py b/py-client/strategy/etf/data.py new file mode 100644 index 0000000..4132751 --- /dev/null +++ b/py-client/strategy/etf/data.py @@ -0,0 +1,55 @@ +"""ETF 专用历史日线适配，不依赖或修改 QMT SDK。""" + +from datetime import date, datetime +import math +import re + +import httpx + + +DAILY_URL = 'http://go.apinb.com/a/get_daily' + + +def daily_bars(client: httpx.Client, code: str, today: date, count: int = 120) -> list[dict]: + """读取指定证券日线；只使用 code 参数，截取历史窗口在本地完成。""" + response = client.get(DAILY_URL, params={'code': code}) + response.raise_for_status() + return parse_daily(response.json(), code, today, count) + + +def parse_daily(payload: dict, code: str, today: date, count: int = 120) -> list[dict]: + """校验业务状态、证券归属和 OHLC，将 trade_date 转为指标需要的 date。""" + if type(count) is not int or count <= 0: + raise ValueError('日线数量必须为正整数') + if not isinstance(payload, dict) or type(payload.get('code')) is not int or payload['code'] != 0: + raise ValueError(f'日线接口业务失败：{payload.get("message", "状态无效") if isinstance(payload, dict) else "响应非对象"}') + details = payload.get('details') + if not isinstance(details, list) or not details: + raise ValueError(f'{code} 日线接口未返回有效 details 列表') + bars = {} + for row in details: + if not isinstance(row, dict) or row.get('ts_code') != code: + raise ValueError(f'{code} 日线证券代码不一致') + stamp = str(row.get('trade_date', '')) + if not re.fullmatch(r'[0-9]{8}', stamp): + raise ValueError(f'{code} 日线日期无效：{stamp}') + day = datetime.strptime(stamp, '%Y%m%d').date() + # 当前日及未来日线均不可用于盘中指标，先过滤再截取最近 count 根。 + if day >= today: + continue + if stamp in bars: + raise ValueError(f'{code} 日线日期重复：{stamp}') + values = {} + for key in ('open', 'high', 'low', 'close'): + value = row.get(key) + if isinstance(value, bool) or not isinstance(value, (int, float, str)): + raise ValueError(f'{code} 日线 {key} 无效') + value = float(value) + if not math.isfinite(value) or value <= 0: + raise ValueError(f'{code} 日线 {key} 非有限正数') + values[key] = value + if not (values['low'] <= values['open'] <= values['high'] + and values['low'] <= values['close'] <= values['high']): + raise ValueError(f'{code} 日线 OHLC 关系异常') + bars[stamp] = dict(date=stamp, **values) + return [bars[stamp] for stamp in sorted(bars)[-count:]] diff --git a/py-client/strategy/etf/engine.py b/py-client/strategy/etf/engine.py new file mode 100644 index 0000000..6bb3e1b --- /dev/null +++ b/py-client/strategy/etf/engine.py @@ -0,0 +1,202 @@ +"""串行 ETF 决策：先核对成交，再止盈，最后低吸并预留本轮资金。""" + +from datetime import datetime +import logging as log +import math + +from libs.grid_take_profit import GridState, GridTrailingTracker +from libs.order import OrderBook, PlaceOrderRequest +from libs.watch import DipWatch +from sdk import OP_BUY, OP_SELL, Portfolio, PositionItem, Tick + +from .config import ETFConfig +from .indicators import Indicators +from .state import Store, SymbolState + + +class Engine: + def __init__(self, client, cfg: ETFConfig, store: Store, min_cash_ratio: float, excluded=()): + if not math.isfinite(min_cash_ratio) or not 0 <= min_cash_ratio <= 1: + raise ValueError('ETF min_cash_ratio 必须在 0 到 1 之间') + self.client, self.cfg, self.store = client, cfg, store + if any(state.pending and code not in cfg.codes for code, state in store.symbols.items()): + raise ValueError('存在已从配置移除的 ETF 待确认委托，请保留该标的直到核对完成') + self.min_cash_ratio, self.excluded = min_cash_ratio, set(excluded) + self.orders = OrderBook(cancel_timeout_sec=120) + self.watch = DipWatch(cfg.watch_seconds, cfg.rebound_pct) + + def fee(self, amount: float) -> float: + return max(self.cfg.min_commission, amount * self.cfg.commission_rate) + + def reconcile(self, code: str, state: SymbolState, position: PositionItem, orders) -> bool: + """必须同时得到终态委托与匹配的持仓快照，才解除本地待确认锁。""" + pending = state.pending + # 柜台清仓记录可能保留旧成本；零持仓应统一视为零成本，避免每轮清空低吸观察。 + current_cost = position.open_price if position.volume > 0 else 0.0 + if pending: + matches = [o for o in orders if o.local_order_id == pending['id'] and o.stock_code == code] + if len(matches) != 1: + log.warning('[ETF待确认] %s 订单=%s 回报缺失或不唯一，暂停该标的', code, pending['id']) + return False + order = matches[0] + status = str(order.order_status) + if status not in {'53', '54', '56', '57'}: + log.info('[ETF待确认] %s 订单=%s 状态=%s 成交=%s', code, pending['id'], status, order.volume_traded) + return False + filled = order.volume_traded + if (order.side != pending['side'] or type(filled) is not int or not 0 <= filled <= pending['volume'] + or (status == '56' and filled != pending['volume'])): + log.warning('[ETF待确认] %s 委托方向或成交数量不一致', code) + return False + expected = pending['base_volume'] + (filled if pending['side'] == 'BUY' else -filled) + if position.volume != expected: + log.warning('[ETF待确认] %s 持仓=%s 预期=%s，等待快照同步', code, position.volume, expected) + return False + if filled and pending['side'] == 'BUY': + if not math.isfinite(order.traded_price) or order.traded_price <= 0: + log.warning('[ETF待确认] %s 缺少实际成交均价', code) + return False + state.last_buy = order.traded_price + if filled: + state.reset_profit() + log.info('[ETF回报] %s 订单=%s 状态=%s 成交=%s 持仓=%s', + code, pending['id'], status, filled, position.volume) + state.pending = {} + # 终态已被快照证实，可清理共用委托簿的短期方向缓存。 + self.orders.busy_cache.delete(f"{pending['side']}-{code}") + elif position.volume != state.volume or not math.isclose(current_cost, state.cost, abs_tol=1e-8): + # 配置内已有仓位一并管理；人工改变仓位时重新建立止盈和加仓基准。 + state.reset_profit() + state.last_buy = position.open_price if position.volume > 0 else 0.0 + self.watch.forget(code) + state.volume, state.cost = position.volume, current_cost + if position.volume == 0: + state.last_buy = 0.0 + state.reset_profit() + return True + + def run(self, portfolio: Portfolio, ticks: dict[str, Tick], indicators: dict[str, Indicators], now: datetime): + # 防重看全账户，自动撤单只针对 ETF 前缀。 + self.orders.refresh(self.client, portfolio.orders, cancel_prefix='ETF-') + assets = portfolio.assets + if not all(math.isfinite(v) and v >= 0 for v in (assets.total, assets.available)): + raise ValueError('账户资金无效') + # 先扣除所有未确认买单，不能等遍历到后面的标的才预留。 + pending_cash = sum(s.pending['reserved'] for s in self.store.symbols.values() + if s.pending.get('side') == 'BUY') + cash = max(0.0, assets.available - assets.total * self.min_cash_ratio - pending_cash) + for code in self.cfg.codes: + state = self.store.get(code) + had_pending = bool(state.pending) + position = portfolio.positions.get(code, PositionItem(stock_code=code)) + try: + if (type(position.volume) is not int or position.volume < 0 + or type(position.can_use_volume) is not int or position.can_use_volume < 0 + or type(position.on_road_volume) is not int or position.on_road_volume < 0 + or not math.isfinite(position.open_price)): + raise ValueError('持仓数量或成本无效') + if not self.reconcile(code, state, position, portfolio.orders): + continue + self.store.save() + if code in self.excluded: + self.watch.forget(code) + continue + if self.orders.busy(code, 'BUY') or self.orders.busy(code, 'SELL'): + continue + if position.on_road_volume > 0 or any( + o.stock_code == code and str(o.order_status) not in {'53', '54', '56', '57'} + for o in portfolio.orders + ): + log.info('[ETF跳过] %s 存在在途份额或未知委托状态', code) + continue + tick, ind = ticks.get(code), indicators.get(code) + if ind is None or not self.fresh_tick(tick, now): + self.watch.forget(code) + log.info('[ETF跳过] %s 日线或实时行情无效/过期', code) + continue + price = round(tick.last_price, 3) + if position.volume > 0 and position.open_price <= 0: + raise ValueError('非空持仓缺少有效成本') + if self.sell(code, state, position, price, ind): + continue + cash -= self.buy(code, state, position, price, ind, cash, now) + except Exception: + # 异常后不允许其他标的重复使用可能已提交的资金。 + if not had_pending and state.pending.get('side') == 'BUY': + cash = max(0, cash - state.pending['reserved']) + log.exception('[ETF异常] %s 本轮跳过', code) + + def fresh_tick(self, tick: Tick | None, now: datetime) -> bool: + if tick is None or not math.isfinite(tick.last_price) or tick.last_price <= 0: + return False + try: + stamp = datetime.strptime(tick.raw['timetag'], '%Y%m%d %H:%M:%S') + return stamp.date() == now.date() and 0 <= (now - stamp).total_seconds() <= self.cfg.max_tick_age_seconds + except (KeyError, TypeError, ValueError): + return False + + def sell(self, code: str, state: SymbolState, position: PositionItem, price: float, ind: Indicators) -> bool: + if position.volume <= 0: + return False + cost = position.open_price + volume = min(position.volume, position.can_use_volume) + volume = volume // 100 * 100 + # 即使 T+1 当天不可卖，也持续记录高位与峰值；翌日可卖时继续判断。 + estimate_volume = volume or position.volume + profit = (price - cost) * estimate_volume + enough_profit = ((price - cost) / cost * 100 >= self.cfg.min_profit_pct + and profit > self.fee(cost * estimate_volume) + self.fee(price * estimate_volume)) + if not state.armed: + if price < max(ind.upper, ind.ma60 + ind.grid, cost + ind.grid) or not enough_profit: + return False + state.armed, state.sell_grid = True, ind.grid + state.peak = math.floor((price - cost) / state.sell_grid) + self.store.save() + log.info('[ETF止盈] %s 高位启动，峰值格=%d 格距=%.3f', code, state.peak, state.sell_grid) + return True + # 通过公开 observe 接口恢复跨日峰值，复用现有网格回撤算法。 + tracker = GridTrailingTracker(1.0) + tracker.observe(code, state.peak) + observation = tracker.observe(code, (price - cost) / state.sell_grid) + state.peak = observation.peak_grid + self.store.save() + if observation.state == GridState.RETREAT and enough_profit and volume > 0: + self.submit(code, state, position, 'SELL', volume, price, 0.0) + # 止盈已启动时不同时补仓，避免同一轮买卖冲突。 + return True + + def buy(self, code: str, state: SymbolState, position: PositionItem, price: float, + ind: Indicators, cash: float, now: datetime) -> float: + volume = self.cfg.buy_hands * 100 + if position.volume + volume > self.cfg.max_hands * 100: + self.watch.forget(code) + return 0.0 + # 首次进入 BOLL 下轨且低于均线一格；加仓须比上次实际买入再低至少一格。 + ceiling = min(ind.ma60, state.last_buy - ind.grid) if state.last_buy else ind.ma60 + entry = min(ind.lower, ind.ma60 - ind.grid, ceiling) + if price > ceiling: + self.watch.forget(code) + return 0.0 + if code not in self.watch.data and price > entry: + return 0.0 + amount = round(price, 3) * volume + reserved = amount + self.fee(amount) + if reserved > cash: + return 0.0 + if not self.watch.triggered('ETF低吸', code, price, now): + return 0.0 + self.submit(code, state, position, 'BUY', volume, price, reserved) + return reserved + + def submit(self, code: str, state: SymbolState, position: PositionItem, side: str, + volume: int, price: float, reserved: float): + order_id = self.orders.new_order_id('ETF', side) + # 先持久化再提交；超时、异常、进程重启均不会丢失未确认的意图。 + state.pending = dict(id=order_id, side=side, volume=volume, + base_volume=position.volume, reserved=reserved) + self.store.save() + request = PlaceOrderRequest(OP_BUY if side == 'BUY' else OP_SELL, + code, volume, order_id, 'etf', price=round(price, 3)) + accepted = self.orders.place(self.client, request) + log.info('[ETF委托] %s %s 数量=%d 限价=%.3f 接口返回=%s 订单=%s，等待柜台核对', + code, side, volume, price, accepted, order_id) diff --git a/py-client/strategy/etf/indicators.py b/py-client/strategy/etf/indicators.py new file mode 100644 index 0000000..170cdcb --- /dev/null +++ b/py-client/strategy/etf/indicators.py @@ -0,0 +1,55 @@ +"""仅用已收盘日线计算指标，避免把盘中未完成的日线混入信号。""" + +from dataclasses import dataclass +from datetime import date, datetime +from decimal import Decimal, ROUND_CEILING +import math +from statistics import fmean, pstdev + +from .config import ETFConfig + + +@dataclass(frozen=True) +class Indicators: + day: str + ma60: float + atr: float + lower: float + middle: float + upper: float + grid: float + + +def calculate(rows: list[dict], today: date, cfg: ETFConfig) -> Indicators: + """MA60 + Wilder ATR + BOLL（总体标准差），格距向上取整到 0.001 元。""" + bars = {} + for row in rows: + day = datetime.strptime(str(row['date']), '%Y%m%d').date() + if day >= today: + continue + if day in bars: + raise ValueError('日线包含重复日期') + high, low, close = (float(row[key]) for key in ('high', 'low', 'close')) + if not all(math.isfinite(v) and v > 0 for v in (high, low, close)) or not low <= close <= high: + raise ValueError('日线价格无效') + bars[day] = (high, low, close) + days = sorted(bars) + if len(days) < max(60, cfg.atr_period + 1, cfg.boll_period): + raise ValueError('已收盘日线不足，至少需要 60 根且能计算 ATR') + # 长期停牌或历史缓存未补齐时不使用过期信号；春节等长假允许 15 个自然日。 + if (today - days[-1]).days > 15: + raise ValueError('最近日线超过 15 个自然日，需补齐行情') + values = [bars[d] for d in days] + closes = [v[2] for v in values] + tr = [max(h - l, abs(h - closes[i - 1]), abs(l - closes[i - 1])) + for i, (h, l, _) in enumerate(values) if i > 0] + n = cfg.atr_period + atr = fmean(tr[:n]) + for value in tr[n:]: + atr = (atr * (n - 1) + value) / n + ma = fmean(closes[-60:]) + window = closes[-cfg.boll_period:] + middle, width = fmean(window), cfg.boll_std * pstdev(window) + raw_grid = max(atr * cfg.atr_multiplier, ma * cfg.min_grid_pct / 100, 0.001) + grid = float(Decimal(str(raw_grid)).quantize(Decimal('0.001'), rounding=ROUND_CEILING)) + return Indicators(days[-1].strftime('%Y%m%d'), ma, atr, middle - width, middle, middle + width, grid) diff --git a/py-client/strategy/etf/state.py b/py-client/strategy/etf/state.py new file mode 100644 index 0000000..1a73f92 --- /dev/null +++ b/py-client/strategy/etf/state.py @@ -0,0 +1,65 @@ +"""保存交易意图与网格基准；实际持仓始终以券商快照为准。""" + +from dataclasses import asdict, dataclass, field +from pathlib import Path +import json +import math + +from libs.lockfile import replace_json + + +@dataclass +class SymbolState: + volume: int = 0 + cost: float = 0.0 + last_buy: float = 0.0 + armed: bool = False + sell_grid: float = 0.0 + peak: int = 0 + pending: dict = field(default_factory=dict) + + def reset_profit(self): + """持仓成本或数量改变后，不沿用上轮止盈峰值。""" + self.armed = False + self.sell_grid = 0.0 + self.peak = 0 + + +class Store: + def __init__(self, path: Path, account: str): + self.path, self.account = path, account + self.symbols: dict[str, SymbolState] = {} + if path.exists(): + try: + raw = json.loads(path.read_text(encoding='utf-8')) + if raw['version'] != 1 or raw['account'] != account: + raise ValueError('版本或账户不一致') + for code, value in raw['symbols'].items(): + state = SymbolState(**value) + if type(state.volume) is not int or state.volume < 0 or type(state.peak) is not int: + raise ValueError('状态数量或峰值无效') + if any(not math.isfinite(v) or v < 0 for v in (state.cost, state.last_buy, state.sell_grid)): + raise ValueError('状态价格无效') + if type(state.armed) is not bool or (state.armed and state.sell_grid <= 0): + raise ValueError('止盈状态无效') + if not isinstance(state.pending, dict): + raise ValueError('委托状态无效') + if state.pending: + p = state.pending + if (p['side'] not in ('BUY', 'SELL') or not p['id'].startswith('ETF-') + or type(p['volume']) is not int or p['volume'] <= 0 + or (p['side'] == 'BUY' and p['volume'] > 1000) + or type(p['base_volume']) is not int or p['base_volume'] < 0 + or not math.isfinite(p['reserved']) or p['reserved'] < 0): + raise ValueError('待确认委托无效') + self.symbols[code] = state + except (ValueError, KeyError, TypeError, AttributeError) as exc: + raise ValueError(f'ETF 状态损坏，禁止自动重建：{path}') from exc + + def get(self, code: str) -> SymbolState: + return self.symbols.setdefault(code, SymbolState()) + + def save(self): + self.path.parent.mkdir(parents=True, exist_ok=True) + replace_json(self.path, dict(version=1, account=self.account, + symbols={k: asdict(v) for k, v in self.symbols.items()})) diff --git a/py-client/tests/test_etf.py b/py-client/tests/test_etf.py new file mode 100644 index 0000000..090e531 --- /dev/null +++ b/py-client/tests/test_etf.py @@ -0,0 +1,329 @@ +"""ETF 离线回归：指标、真实防飞刀/网格算法、限仓、回报和持久化。""" + +from datetime import date, datetime, timedelta +import httpx +from pathlib import Path +import tempfile +import unittest +from unittest.mock import Mock + +from sdk import Assets, OrderItem, Portfolio, PositionItem, Tick +from strategy.etf.config import ETFConfig, load +from strategy.etf.data import DAILY_URL, daily_bars, parse_daily +from strategy.etf.engine import Engine +from strategy.etf.indicators import Indicators, calculate +from strategy.etf.state import Store + + +CODE, OTHER = '510300.SH', '159915.SZ' +NOW = datetime(2026, 9, 16, 10) +IND = Indicators('20260915', 10, 0.2, 9.5, 10, 10.5, 0.2) + + +def tick(price, now=NOW): + return Tick(price, raw={'timetag': now.strftime('%Y%m%d %H:%M:%S')}) + + +def position(volume=0, cost=0, available=None): + return PositionItem(stock_code=CODE, volume=volume, open_price=cost, + can_use_volume=volume if available is None else available) + + +class ETFTests(unittest.TestCase): + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.path = Path(temp.name) / 'state.json' + self.client = Mock() + self.client.passorder.return_value = {'status': 'success'} + self.cfg = ETFConfig(codes=(CODE,), min_commission=0, commission_rate=0) + self.store = Store(self.path, 'test') + self.engine = Engine(self.client, self.cfg, self.store, 0.1) + + def run_price(self, price, pos=None, orders=(), cash=10000, now=NOW, ind=IND): + portfolio = Portfolio(Assets(total=10000, available=cash), {CODE: pos or position()}, list(orders)) + self.engine.run(portfolio, {CODE: tick(price, now)}, {CODE: ind}, now) + + def buy(self): + self.run_price(9.4) + self.run_price(9.46) + self.client.passorder.assert_called_once() + + def report(self, status=56, filled=100, side=23, price=9.46): + pending = self.store.get(CODE).pending + return OrderItem(stock_code=CODE, remark=pending['id'] + '|etf', offset_flag=side, + volume_traded=filled, volume_total_original=pending['volume'], + traded_price=price, order_status=status) + + def test_boll_lower_requires_rebound_and_uses_fixed_limit_order(self): + self.run_price(9.8) + self.run_price(9.4) + self.run_price(9.3) + self.run_price(9.35) + self.client.passorder.assert_not_called() + self.run_price(9.36) + request = self.client.passorder.call_args.kwargs + self.assertEqual((request['volume'], request['price'], request['pr_type']), (100, 9.36, 11)) + self.assertEqual(request['strategy_name'], 'etf') + self.assertTrue(self.store.get(CODE).pending) + + def test_pending_written_before_network_and_retained_after_timeout(self): + def submit(**kwargs): + saved = Store(self.path, 'test').get(CODE).pending + self.assertEqual(saved['id'], kwargs['order_id']) + raise TimeoutError('unknown result') + self.client.passorder.side_effect = submit + self.run_price(9.4) + with self.assertLogs(level='ERROR'): + self.run_price(9.46) + self.engine = Engine(self.client, self.cfg, Store(self.path, 'test'), 0.1) + with self.assertLogs(level='WARNING'): + self.run_price(9.3, now=NOW + timedelta(minutes=10)) + self.client.passorder.assert_called_once() + + def test_filled_order_waits_for_position_snapshot(self): + self.buy() + report = self.report() + with self.assertLogs(level='WARNING'): + self.run_price(9.2, orders=[report]) + self.assertTrue(self.store.get(CODE).pending) + self.run_price(9.2, position(100, 9.46, 0), [report]) + self.assertFalse(self.store.get(CODE).pending) + self.assertEqual(self.store.get(CODE).last_buy, 9.46) + self.client.passorder.assert_called_once() + + def test_add_requires_another_grid_below_actual_fill(self): + self.buy() + report = self.report() + pos = position(100, 9.46, 0) + self.run_price(9.4, pos, [report]) + self.run_price(9.46, pos) + self.client.passorder.assert_called_once() + self.run_price(9.1, pos) + self.run_price(9.16, pos) + self.assertEqual(self.client.passorder.call_count, 2) + + def test_partial_cancel_records_actual_fill_and_never_exceeds_cap(self): + self.cfg = ETFConfig(codes=(CODE,), buy_hands=2, min_commission=0, commission_rate=0) + self.engine = Engine(self.client, self.cfg, self.store, 0.1) + pos = position(800, 10) + self.run_price(9.4, pos) + self.run_price(9.46, pos) + report = self.report(status=53, filled=100) + self.run_price(9.1, position(900, 9.94), [report]) + self.run_price(9.16, position(900, 9.94)) + self.client.passorder.assert_called_once() + self.assertFalse(self.store.get(CODE).pending) + + def test_full_position_blocks_buy_and_zero_position_is_not_a_warning(self): + self.run_price(9.4, position(1000, 10)) + self.run_price(9.46, position(1000, 10)) + self.client.passorder.assert_not_called() + with self.assertNoLogs(level='WARNING'): + self.run_price(9.8, position()) + + def test_zero_position_with_retained_broker_cost_can_reopen(self): + self.run_price(9.4, position(0, 10)) + self.run_price(9.46, position(0, 10)) + self.client.passorder.assert_called_once() + + def test_rejected_order_is_logged_and_does_not_advance_anchor(self): + self.buy() + report = self.report(status=57, filled=0) + self.run_price(9.8, orders=[report]) + self.assertEqual(self.store.get(CODE).last_buy, 0) + self.assertFalse(self.store.get(CODE).pending) + + def test_t_plus_one_tracks_peak_but_only_sells_available_whole_lots(self): + self.run_price(10.7, position(200, 10, 0)) + self.run_price(10.55, position(200, 10, 0)) + self.client.passorder.assert_not_called() + self.engine = Engine(self.client, self.cfg, Store(self.path, 'test'), 0.1) + self.run_price(10.55, position(200, 10, 100)) + order = self.client.passorder.call_args.kwargs + self.assertEqual((order['op_type'], order['volume']), (24, 100)) + + def test_drop_below_activation_price_still_triggers_profitable_retreat(self): + self.run_price(10.7, position(100, 10)) + self.run_price(10.4, position(100, 10)) + self.assertEqual(self.client.passorder.call_args.kwargs['op_type'], 24) + + def test_cost_change_and_flat_position_reset_peak(self): + self.run_price(10.7, position(100, 10)) + self.assertTrue(self.store.get(CODE).armed) + self.run_price(10.5, position(200, 10.4)) + self.assertFalse(self.store.get(CODE).armed) + self.run_price(9.8, position()) + self.assertEqual(self.store.get(CODE).last_buy, 0) + self.client.passorder.assert_not_called() + + def test_fee_floor_prevents_loss_after_commission(self): + cfg = ETFConfig(codes=(CODE,), min_commission=50, commission_rate=0) + self.engine = Engine(self.client, cfg, self.store, 0.1) + self.run_price(10.7, position(100, 10)) + self.run_price(10.55, position(100, 10)) + self.client.passorder.assert_not_called() + + def test_cash_reserve_and_fixed_lot_no_downsizing(self): + self.run_price(9.4, cash=1900) + self.run_price(9.46, cash=1900) + self.client.passorder.assert_not_called() + + def test_multiple_symbols_share_one_cash_budget(self): + cfg = ETFConfig(codes=(CODE, OTHER), min_commission=0, commission_rate=0) + self.engine = Engine(self.client, cfg, self.store, 0.1) + portfolio = Portfolio(Assets(total=10000, available=2500), {}, []) + for price in (9.4, 9.46): + self.engine.run(portfolio, {c: tick(price) for c in cfg.codes}, {c: IND for c in cfg.codes}, NOW) + self.client.passorder.assert_called_once() + + def test_other_strategy_order_blocks_same_symbol_without_cancel(self): + report = OrderItem(stock_code=CODE, remark='TREN-BUY-other', offset_flag=23, + order_status=50, insert_date='20260916', insert_time='093000') + self.run_price(9.4, orders=[report]) + self.run_price(9.46, orders=[report]) + self.client.passorder.assert_not_called() + self.client.cancel_by_id.assert_not_called() + + def test_pending_later_symbol_reserves_cash_before_first_symbol(self): + cfg = ETFConfig(codes=(CODE, OTHER), min_commission=0, commission_rate=0) + self.store.get(OTHER).pending = dict(id='ETF-BUY-pending', side='BUY', volume=100, + base_volume=0, reserved=950) + self.engine = Engine(self.client, cfg, self.store, 0.1) + portfolio = Portfolio(Assets(total=10000, available=2500), {}, []) + with self.assertLogs(level='WARNING'): + for price in (9.4, 9.46): + self.engine.run(portfolio, {CODE: tick(price)}, {CODE: IND}, NOW) + self.client.passorder.assert_not_called() + + def test_on_road_or_unknown_order_never_opens_another_buy(self): + pos = position() + pos.on_road_volume = 100 + self.run_price(9.4, pos) + self.run_price(9.46, pos) + unknown = OrderItem(stock_code=CODE, order_status=255) + self.run_price(9.4, orders=[unknown]) + self.run_price(9.46, orders=[unknown]) + self.client.passorder.assert_not_called() + + def test_excluded_symbol_is_neither_bought_nor_sold(self): + self.engine.excluded.add(CODE) + self.run_price(9.4) + self.run_price(9.46) + self.run_price(10.7, position(100, 10)) + self.run_price(10.55, position(100, 10)) + self.client.passorder.assert_not_called() + + def test_invalid_or_stale_tick_cannot_trade(self): + for t in (None, Tick(10), tick(float('nan')), tick(9.4, NOW - timedelta(days=1)), + tick(9.4, NOW - timedelta(seconds=91))): + self.assertFalse(self.engine.fresh_tick(t, NOW)) + self.assertTrue(self.engine.fresh_tick(tick(9.4), NOW)) + + def test_corrupt_state_does_not_silently_start_empty(self): + self.path.write_text('{', encoding='utf-8') + with self.assertRaises(ValueError): + Store(self.path, 'test') + + +class IndicatorTests(unittest.TestCase): + def bars(self): + days = [] + day = date(2026, 9, 15) + while len(days) < 80: + if day.weekday() < 5: + days.append(day) + day -= timedelta(days=1) + return [dict(date=d.strftime('%Y%m%d'), high=11, low=9, close=10) for d in reversed(days)] + + def test_known_constant_series_and_exclusion_of_unfinished_day(self): + cfg = ETFConfig(codes=(CODE,)) + rows = self.bars() + [dict(date='20260916', high=999, low=1, close=999)] + ind = calculate(rows, NOW.date(), cfg) + self.assertEqual((ind.ma60, ind.atr, ind.lower, ind.upper, ind.grid), (10, 2, 10, 10, 2)) + + def test_atr_accounts_for_gap_and_uses_wilder_smoothing(self): + rows = self.bars() + rows[-1].update(high=14, low=12, close=13) + ind = calculate(rows, NOW.date(), ETFConfig(codes=(CODE,))) + self.assertAlmostEqual(ind.atr, (2 * 13 + 4) / 14) + self.assertAlmostEqual(ind.ma60, 10.05) + self.assertGreater(ind.upper, ind.middle) + + def test_bad_or_insufficient_history_is_rejected(self): + cfg = ETFConfig(codes=(CODE,)) + for rows in (self.bars()[:59], self.bars() + [self.bars()[-1]], + self.bars()[:-1] + [dict(self.bars()[-1], close=float('nan'))]): + with self.assertRaises(ValueError): + calculate(rows, NOW.date(), cfg) + + def test_grid_floor_and_tick_rounding(self): + rows = [dict(row, high=10.001, low=9.999) for row in self.bars()] + ind = calculate(rows, NOW.date(), ETFConfig(codes=(CODE,), min_grid_pct=0.501)) + self.assertEqual(ind.grid, 0.051) + + +class ConfigAndDataTests(unittest.TestCase): + def test_config_rejects_excess_hands_and_invalid_codes(self): + for kwargs in ({'max_hands': 11}, {'buy_hands': 11}, {'buy_hands': True}, + {'atr_multiplier': float('nan')}, {'codes': ('920202.BJ',)}, + {'codes': (CODE, CODE)}, {'codes': ()}): + with self.assertRaises(ValueError): + ETFConfig(**dict({'codes': (CODE,)}, **kwargs)) + + def test_default_file_loads(self): + cfg = load() + self.assertEqual((cfg.buy_hands, cfg.max_hands), (1, 10)) + + + +class DailyDataTests(unittest.TestCase): + def row(self, day=20260915, **changes): + return dict(dict(ts_code=CODE, trade_date=day, open=10, high=11, low=9, close=10), **changes) + + def test_request_and_sample_shape(self): + def respond(request): + self.assertEqual(str(request.url), DAILY_URL + '?code=' + CODE) + self.assertNotIn('x-token', request.headers) + return httpx.Response(200, json={'code': 0, 'message': '', 'details': [self.row()]}) + with httpx.Client(transport=httpx.MockTransport(respond)) as client: + self.assertEqual(daily_bars(client, CODE, NOW.date()), + [dict(date='20260915', open=10.0, high=11.0, low=9.0, close=10.0)]) + + def test_sort_filter_then_limit_and_numeric_strings(self): + payload = dict(code=0, details=[self.row(20260916), self.row(20260915, close='10.5'), + self.row(20260914), self.row(20260917)]) + bars = parse_daily(payload, CODE, NOW.date(), count=1) + self.assertEqual([b['date'] for b in bars], ['20260915']) + self.assertEqual(bars[0]['close'], 10.5) + + def test_http_error_and_invalid_json_propagate(self): + for status, content in ((404, '{}'), (200, 'error')): + with httpx.Client(transport=httpx.MockTransport( + lambda r: httpx.Response(status, text=content))) as client: + with self.assertRaises((httpx.HTTPStatusError, ValueError)): + daily_bars(client, CODE, NOW.date()) + + def test_bad_business_response_is_rejected(self): + for payload in (None, [], {}, {'code': False, 'details': [self.row()]}, + {'code': 1, 'message': 'failed'}, {'code': 0, 'details': []}, + {'code': 0, 'details': {}}, {'code': 0, 'details': None}): + with self.subTest(payload=payload), self.assertRaises(ValueError): + parse_daily(payload, CODE, NOW.date()) + + def test_wrong_symbol_duplicate_dates_and_invalid_ohlc_are_rejected(self): + for rows in ([self.row(ts_code=OTHER)], [self.row(), self.row()], + [self.row(20260230)], [self.row(close=float('nan'))], + [self.row(open=True)], [self.row(low=12)], [self.row(close=None)]): + with self.subTest(rows=rows), self.assertRaises(ValueError): + parse_daily(dict(code=0, details=rows), CODE, NOW.date()) + + def test_external_history_flows_into_real_indicators(self): + details = [self.row(int(row['date'])) for row in IndicatorTests().bars()] + rows = parse_daily(dict(code=0, details=details), CODE, NOW.date()) + ind = calculate(rows, NOW.date(), ETFConfig(codes=(CODE,))) + self.assertEqual((ind.ma60, ind.atr, ind.grid), (10, 2, 2)) + + +if __name__ == '__main__': + unittest.main()