44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
"""Phase 2 端到端冒烟：result_store + 因子 registry + API + data_adapter（真实数据）。"""
|
|
import sys
|
|
sys.path.insert(0, "/app")
|
|
sys.path.insert(0, "/app/vnpy_v4.4.0")
|
|
|
|
|
|
def main():
|
|
print("=== Phase 2 E2E Smoke ===")
|
|
import pandas as pd
|
|
|
|
# 1. result_store 增删查
|
|
from sanguo_backtest.result_store import save_result, load_result, BacktestResult
|
|
r = BacktestResult("smoke", "cta", "done", "Test", "600000", {"n": 5},
|
|
"2024-01-01", "2024-06-30", {"return": 0.1},
|
|
pd.DataFrame({"e": [1.0]}), pd.DataFrame(), None)
|
|
rid = save_result(r, db_path="/tmp/smoke.db", file_dir="/tmp")
|
|
print(f" result_store: id={rid}, strategy={load_result(rid, '/tmp/smoke.db').strategy}")
|
|
|
|
# 2. 因子 registry
|
|
import sanguo_factor.library
|
|
from sanguo_factor.registry import list_factors
|
|
print(f" factors: {len(list_factors())} total, {len(list_factors('builtin'))} builtin")
|
|
|
|
# 3. API 启动
|
|
from sanguo_api.app import create_app
|
|
app = create_app(db_path="/tmp/smoke.db", file_dir="/tmp")
|
|
routes = [r.path for r in app.routes if hasattr(r, "path")]
|
|
print(f" API routes: {len(routes)} loaded")
|
|
|
|
# 4. data_adapter（真实 A 股数据）
|
|
from sanguo_data.datareader import read_db_daily
|
|
from sanguo_data.config import load_config
|
|
from sanguo_factor.data_adapter import convert_bars_to_alpha_df
|
|
cfg = load_config("/app/config/data_platform.yaml")
|
|
bars = read_db_daily("600000", "2024-01-01", "2024-06-30", cfg)
|
|
df = convert_bars_to_alpha_df(bars)
|
|
print(f" data_adapter: {len(bars)} bars -> df height={df.height}, cols={df.columns[:5]}...")
|
|
|
|
print("=== Phase 2 Smoke DONE ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|