claude_dev
164690373f
- 资金占用成本(spec§195): StrategyRunner.daily_borrow_cost(used×risk_free/365) 归因per_strategy_pnl(不碰account总账, account.equity真实净值不变); config risk_free_rate=0.02; engine.step mark_to_market后计扣; =0向后兼容跳过 - 分红送股(spec§295): dividend_source.py(akshare stock_history_dividend_detail, 实测600000/000001纯现金分红); PositionLedger.apply_split(volume×factor/avg÷factor); Account.apply_cash_dividend; engine._apply_dividends(除权日调整,现金先split后); mark_to_market停牌prev_close兜底(今收→前收→均价); _run_replay注入dividends日历 - 修_restore_ledger预存bug: PositionLedger.__init__加volume/frozen/avg_price参数 (原只symbol, live_orchestrator跨日恢复4参数调用会TypeError, 首次step空仓未暴露) - 139 passed(119基准+20分红+3占用成本), 无回归 - live_step dividends注入待分期项(每日拉全市场分红慢, 需run_daily_update预拉日历)
46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
# sanguo_data/config.py
|
|
from dataclasses import dataclass
|
|
import os
|
|
import yaml
|
|
|
|
@dataclass(frozen=True)
|
|
class DataConfig:
|
|
data_paths: dict
|
|
data_sources: dict
|
|
validation: dict
|
|
performance: dict
|
|
risk_free_rate: float = 0.02 # 年化无风险利率（spec §195 资金占用成本归因）
|
|
|
|
def load_config(path: str) -> DataConfig:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
except FileNotFoundError:
|
|
raise FileNotFoundError(f"配置文件不存在: {path}")
|
|
except yaml.YAMLError as e:
|
|
raise ValueError(f"YAML解析失败: {e}")
|
|
|
|
if not raw:
|
|
raise ValueError(f"配置文件为空: {path}")
|
|
|
|
return DataConfig(
|
|
data_paths=raw.get("data_paths", {}),
|
|
data_sources=raw.get("data_sources", {}),
|
|
validation=raw.get("validation", {}),
|
|
performance=raw.get("performance", {}),
|
|
risk_free_rate=float(raw.get("risk_free_rate", 0.02)),
|
|
)
|
|
|
|
|
|
def find_config_path() -> str:
|
|
"""Locate data_platform.yaml: container /app/config first, then repo config/."""
|
|
candidates = [
|
|
"/app/config/data_platform.yaml",
|
|
os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "data_platform.yaml"),
|
|
"config/data_platform.yaml",
|
|
]
|
|
for p in candidates:
|
|
if os.path.exists(p):
|
|
return p
|
|
return candidates[0]
|