66 lines
2.9 KiB
Python
66 lines
2.9 KiB
Python
"""保存交易意图与网格基准；实际持仓始终以券商快照为准。"""
|
|
|
|
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()}))
|