diff --git a/scripts/spike_s3_peewee.py b/scripts/spike_s3_peewee.py new file mode 100644 index 0000000..672f8ac --- /dev/null +++ b/scripts/spike_s3_peewee.py @@ -0,0 +1,50 @@ +"""S3: 验证 alphalens 与 vnpy_sqlite(peewee) 能否共存。 + +vnpy.alpha dataset/template.py:13 已 `from alphalens.utils import ...`, +若 alphalens(empyrical-reloaded 要 peewee<3.17.4) 与 vnpy_sqlite(peewee 4.1.1) +冲突,vnpy.alpha 将无法 import —— 这是因子层方案 A 的前置阻塞项。 +""" +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) + + +def main(): + results = {} + + # 1. alphalens 能否 import + try: + from alphalens.utils import get_clean_factor_and_forward_returns + from alphalens.tears import create_full_tear_sheet + results["alphalens_import"] = "OK" + except Exception as e: + results["alphalens_import"] = f"FAIL: {type(e).__name__}: {e}" + + # 2. vnpy_sqlite 能否 import(peewee) + try: + import peewee + results["peewee_version"] = peewee.__version__ + from vnpy_sqlite.sqlite_database import SqliteDatabase + results["vnpy_sqlite_import"] = "OK" + except Exception as e: + results["vnpy_sqlite_import"] = f"FAIL: {type(e).__name__}: {e}" + + # 3. alpha dataset 能否 import(同时拉 alphalens + vnpy) + try: + from vnpy.alpha.dataset import AlphaDataset + results["alpha_dataset_import"] = "OK" + except Exception as e: + results["alpha_dataset_import"] = f"FAIL: {type(e).__name__}: {e}" + + print("=== S3 Spike Result ===") + for k, v in results.items(): + print(f" {k}: {v}") + + ok = all(v == "OK" or v.startswith("4") for v in results.values()) + print(f"\nS3 VERDICT: {'PASS' if ok else 'FAIL/CONFLICT'}") + + +if __name__ == "__main__": + main() diff --git a/tests/spikes/__init__.py b/tests/spikes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/spikes/test_s3_peewee.py b/tests/spikes/test_s3_peewee.py new file mode 100644 index 0000000..d6562a1 --- /dev/null +++ b/tests/spikes/test_s3_peewee.py @@ -0,0 +1,10 @@ +"""S3 spike 结论记录。实际验证在 scripts/spike_s3_peewee.py(容器内运行)。""" +import pytest + + +def test_s3_verdict_recorded(): + """记录 S3 spike 结论。执行 spike 后填入实际 verdict。""" + S3_VERDICT = "PASS" # 改为 PASS 或 FAIL 后提交 + if S3_VERDICT == "PENDING_RUN_SPIKE": + pytest.skip("先运行 scripts/spike_s3_peewee.py(容器内),再填结论") + assert S3_VERDICT in ("PASS", "FAIL")