claude_dev
510f77e6ea
- BacktestResult 加 id；save_result 设 result.id=lastrowid（修 get_result bug） - runner._on_done 用 result.id（getattr 兜底 FactorReport） - cta_engine 构建 equity_curve/trades DataFrame；save 传 file_dir - result_store parquet→JSON（去 pyarrow 依赖，本机/容器都稳） - 16 tests passed
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Quick test to verify the real tears pipeline runs in container."""
|
|
import sys
|
|
import os
|
|
_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "vnpy_v4.4.0"))
|
|
sys.path.insert(0, _VNPY_SRC)
|
|
|
|
from unittest.mock import Mock, MagicMock
|
|
from sanguo_factor.analyzer import run_factor_analysis
|
|
from sanguo_factor.registry import register_factor
|
|
|
|
# Register a simple test factor
|
|
register_factor("test_ma5", "ts_mean(close, 5)")
|
|
|
|
# Create minimal cfg mock
|
|
cfg = Mock()
|
|
cfg.data_paths = {"vnpy_db": "/tmp/test.db"}
|
|
|
|
try:
|
|
# Run with minimal data
|
|
result = run_factor_analysis(
|
|
symbols=["600000.SH"], # Single symbol
|
|
factor_names=["test_ma5"],
|
|
start="2024-01-01",
|
|
end="2024-01-31", # Small date range
|
|
cfg=cfg,
|
|
output_dir="/tmp/test_tears"
|
|
)
|
|
|
|
print(f"Test completed successfully!")
|
|
print(f"Result: {result}")
|
|
print(f"IC Summary: {result.ic_summary}")
|
|
print(f"Report Path: {result.report_path}")
|
|
|
|
except Exception as e:
|
|
print(f"Test failed with error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|