60 lines
2.8 KiB
Python
60 lines
2.8 KiB
Python
"""独立读取 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)
|