claude_dev
28aea67232
- 修 submit_cta/optimize 策略字符串→类解析（get_strategy_class） - cta_engine: worker 进程设 vnpy DB→quant_trading.db（修 0 根数据） - equity_curve 取自 calculate_result 的 daily_df（修 get_all_daily_results 对象问题） - kline 补 cfg（find_config_path 共享） - 端到端冒烟通过：DoubleMaStrategy 600000 → equity111/pnl111/trades1/kline117
44 lines
1.2 KiB
Python
44 lines
1.2 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
|
|
|
|
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", {}),
|
|
)
|
|
|
|
|
|
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]
|