claude_dev
3ef8c617f0
- sanguo_data/config.py: load_config 加 env 覆盖(SANGUO_DATA_ROOT 重映射 daily/raw/qfq/15min/vnpy_db;SANGUO_LIVE_ENABLED/SANGUO_BRIDGE_URL 覆盖 live 段), 三机共用一份 git config 零漂移 - sanguo_api/main.py: build_app 加 SANGUO_DB_PATH/SANGUO_FILE_PATH env 覆盖 - docs/deployment/vps-native-brain.md: VPS原生部署全记录(安装/配置/服务/验证/ 数据staging/Phase2暂缓理由) 实证:VPS vnpy4.4.0+vnpy_qmt0.3.3 连真实miniQMT读真实账户/持仓; live_step全链路+影子transport(本机bridge)通过。详见文档。
71 lines
2.7 KiB
Python
71 lines
2.7 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 资金占用成本归因）
|
|
live: dict | None = None # 实盘集成（D期 spec §5），None/缺省=disabled
|
|
|
|
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}")
|
|
|
|
data_paths = dict(raw.get("data_paths", {}))
|
|
# 部署环境路径覆盖（VPS native 等）：SANGUO_DATA_ROOT 设定时，*_dir 与 vnpy_db
|
|
# 重映射到该根下，保持 datareader 期望的叶子结构（daily/raw/qfq/15min*/quant_trading.db）。
|
|
# 不设则用 yaml 原值（NAS/Mac）。避免改部署态配置文件，保 git 真相。
|
|
root = os.environ.get("SANGUO_DATA_ROOT")
|
|
if root:
|
|
_leaf = {"daily_dir": "daily", "raw_dir": "raw", "qfq_dir": "qfq",
|
|
"minute_15_dir": "15min", "minute_15_qfq_dir": "15min_qfq",
|
|
"minute_15_raw_dir": "15min_raw"}
|
|
for _k, _suf in _leaf.items():
|
|
if _k in data_paths:
|
|
data_paths[_k] = os.path.join(root, _suf)
|
|
if "vnpy_db" in data_paths:
|
|
data_paths["vnpy_db"] = os.path.join(root, "quant_trading.db")
|
|
|
|
live = raw.get("live")
|
|
if isinstance(live, dict):
|
|
live = dict(live)
|
|
if os.environ.get("SANGUO_LIVE_ENABLED"):
|
|
live["enabled"] = os.environ["SANGUO_LIVE_ENABLED"].strip().lower() in ("1", "true", "yes", "on")
|
|
if os.environ.get("SANGUO_BRIDGE_URL"):
|
|
live["bridge_url"] = os.environ["SANGUO_BRIDGE_URL"].strip()
|
|
|
|
return DataConfig(
|
|
data_paths=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)),
|
|
live=live,
|
|
)
|
|
|
|
|
|
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]
|