File size: 3,654 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | #!/usr/bin/env python3
"""
Full qlib-native research pipeline:
1. (Optional) Parquet/CSV → qlib .bin dump
2. GP factor mining on qlib data
3. Export GP features → qlib DataHandler/DatasetH pickles
4. Run qlib experiments (Alpha158 baseline + GP LightGBM)
5. Generate professional research reports
"""
import argparse
import os
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from config.settings import load_settings
from factor_engine.gp.mining import run_gp_mining
def _run(cmd: list[str], env: dict | None = None):
print(f"\n>>> {' '.join(cmd)}")
subprocess.run(cmd, check=True, env=env or os.environ.copy())
def main():
parser = argparse.ArgumentParser(description="End-to-end qlib research pipeline")
parser.add_argument("--run-id", type=str, default=None, help="Experiment run id")
parser.add_argument("--skip-dump", action="store_true", help="Skip parquet→bin conversion")
parser.add_argument("--skip-gp", action="store_true", help="Skip GP mining (reuse existing outputs)")
parser.add_argument("--skip-qrun", action="store_true", help="Skip qlib qrun experiments")
parser.add_argument("--parquet", type=str, default=None, help="Source parquet for bin dump")
parser.add_argument("--baseline-only", action="store_true", help="Only run Alpha158 baseline")
args = parser.parse_args()
settings = load_settings()
run_id = args.run_id or os.environ.get("RUN_ID", "qlib_gp_run_0")
env = os.environ.copy()
env["RUN_ID"] = run_id
env.setdefault("MLFLOW_ALLOW_FILE_STORE", "true")
py = sys.executable
print("=" * 70)
print("ML-Alpha-Research-System — Qlib Native Pipeline")
print(f"RUN_ID={run_id}")
print("=" * 70)
if not args.skip_dump and args.parquet:
print("\n[Step 1/5] Dump raw data to qlib .bin format")
_run([py, str(ROOT / "data_pipeline" / "convert_to_qlib_bin.py"), "--parquet", args.parquet], env)
else:
print("\n[Step 1/5] Skip bin dump (using existing qlib data or --skip-dump)")
if not args.skip_gp and not args.baseline_only:
print("\n[Step 2/5] GP factor mining")
run_gp_mining()
else:
print("\n[Step 2/5] Skip GP mining")
if not args.baseline_only:
print("\n[Step 3/5] Build GP qlib dataset/handler artifacts")
_run([py, str(ROOT / "scripts" / "build_gp_dataset.py"), "--run-id", run_id], env)
if not args.skip_qrun:
print("\n[Step 4/5] Run qlib experiments (Recorder + mlruns)")
_run(
[py, str(ROOT / "scripts" / "run_qrun.py"), "--config", "config/workflows/workflow_alpha158_baseline.yaml"],
env,
)
if not args.baseline_only:
_run(
[
py,
str(ROOT / "scripts" / "run_qrun.py"),
"--config",
"config/workflows/workflow_gp_lightgbm.yaml",
"--run-id",
run_id,
],
env,
)
else:
print("\n[Step 4/5] Skip qlib experiments")
print("\n[Step 5/5] Generate reports")
_run([py, str(ROOT / "scripts" / "generate_report.py"), "--run-id", run_id], env)
print("\n" + "=" * 70)
print("Pipeline complete.")
print(f" GP outputs: {settings.gp_output_dir(run_id)}")
print(f" mlruns: {settings.mlruns_uri}")
print(f" reports: {settings.path(settings.raw['output']['reports_dir'])}")
print("=" * 70)
if __name__ == "__main__":
main()
|