diff --git a/.github/workflows/publish-lifecycle-inputs.yml b/.github/workflows/publish-lifecycle-inputs.yml new file mode 100644 index 0000000..cf9094a --- /dev/null +++ b/.github/workflows/publish-lifecycle-inputs.yml @@ -0,0 +1,61 @@ +name: Publish Lifecycle Inputs + +on: + schedule: + - cron: "0 4 * * 1" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.ref_name }} + cancel-in-progress: false + +jobs: + publish: + if: github.ref == format('refs/heads/{0}', github.event.repository.default_branch) + runs-on: [self-hosted, Linux, X64] + timeout-minutes: 60 + env: + DOWNLOAD_TOP_LIQUID: "90" + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + set -euo pipefail + python -m pip install --upgrade pip + requirements_file=requirements-lock.txt + if [ ! -f "${requirements_file}" ]; then requirements_file=requirements.txt; fi + python -m pip install -r "${requirements_file}" + + - name: Download real Binance history + run: | + set -euo pipefail + completed_date="$(python -c 'from datetime import datetime, timedelta, timezone; print((datetime.now(timezone.utc) - timedelta(days=1)).date())')" + python scripts/download_history.py --top-liquid "${DOWNLOAD_TOP_LIQUID}" \ + --end-date "${completed_date}" \ + --force-exchange-info + python scripts/download_history.py --symbols ETHUSDT --end-date "${completed_date}" + + - name: Export production research inputs + run: | + set -euo pipefail + python scripts/export_lifecycle_preflight_inputs.py \ + --universe-mode broad_liquid \ + --output-dir "${RUNNER_TEMP}/crypto-lifecycle-inputs" + + - name: Upload lifecycle inputs + uses: actions/upload-artifact@v7 + with: + name: crypto-lifecycle-inputs-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/crypto-lifecycle-inputs + if-no-files-found: error + retention-days: 30 diff --git a/scripts/export_lifecycle_preflight_inputs.py b/scripts/export_lifecycle_preflight_inputs.py new file mode 100644 index 0000000..5b1a17e --- /dev/null +++ b/scripts/export_lifecycle_preflight_inputs.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from src.config import load_config +from src.pipeline import run_research_pipeline + +PANEL_COLUMNS = ("in_universe", "open", "final_score") +COMBO_SYMBOLS = ("BTCUSDT", "ETHUSDT") +MIN_PANEL_DAYS = 730 +MIN_MARKET_DAYS = 900 +MAX_FRESHNESS_DAYS = 3 + + +def export_lifecycle_inputs(panel: pd.DataFrame, output_dir: Path) -> dict[str, object]: + if list(panel.index.names) != ["date", "symbol"]: + raise ValueError("research panel must use a date/symbol MultiIndex") + missing = sorted(set((*PANEL_COLUMNS, "close")) - set(panel.columns)) + if missing: + raise ValueError(f"research panel is missing columns: {', '.join(missing)}") + + frame = panel.reset_index().copy() + frame["date"] = pd.to_datetime(frame["date"], errors="coerce").dt.tz_localize(None).dt.normalize() + frame["symbol"] = frame["symbol"].astype(str).str.strip().str.upper() + today = pd.Timestamp.now(tz="UTC").tz_localize(None).normalize() + frame = frame.loc[frame["date"] < today] + lifecycle_panel = frame[["date", "symbol", *PANEL_COLUMNS]].dropna(subset=["date", "open"]) + scored_panel = lifecycle_panel.dropna(subset=["final_score"]) + if scored_panel.empty: + raise ValueError("research panel has no scored lifecycle rows") + panel_dates = pd.DatetimeIndex(sorted(scored_panel["date"].unique())) + if len(panel_dates) < MIN_PANEL_DAYS: + raise ValueError(f"research panel requires at least {MIN_PANEL_DAYS} scored dates") + in_universe_counts = scored_panel.loc[scored_panel["in_universe"]].groupby("date")["symbol"].nunique() + in_universe_counts = in_universe_counts.reindex(panel_dates, fill_value=0) + if int(in_universe_counts.min()) < 2: + raise ValueError("research panel requires at least two in-universe symbols per scored date") + + market_history = frame.loc[frame["symbol"].isin(COMBO_SYMBOLS), ["date", "symbol", "close"]].dropna() + missing_combo = sorted(set(COMBO_SYMBOLS) - set(market_history["symbol"])) + if missing_combo: + raise ValueError(f"research panel is missing combo symbols: {', '.join(missing_combo)}") + reference_dates = set(market_history.loc[market_history["symbol"] == "BTCUSDT", "date"]) + if len(reference_dates) < MIN_MARKET_DAYS: + raise ValueError(f"BTC market history requires at least {MIN_MARKET_DAYS} dates") + for symbol in COMBO_SYMBOLS: + symbol_dates = set(market_history.loc[market_history["symbol"] == symbol, "date"]) + if ( + len(symbol_dates & reference_dates) / len(reference_dates) < 0.99 + or min(symbol_dates) > min(reference_dates) + or max(symbol_dates) < max(reference_dates) + ): + raise ValueError(f"market history has incomplete symbol coverage: {symbol}") + if today - panel_dates.max() > pd.Timedelta(days=MAX_FRESHNESS_DAYS): + raise ValueError("research panel is stale") + if today - max(reference_dates) > pd.Timedelta(days=MAX_FRESHNESS_DAYS): + raise ValueError("BTC/ETH market history is stale") + + output_dir.mkdir(parents=True, exist_ok=True) + panel_path = output_dir / "research_panel.csv.gz" + market_path = output_dir / "market_history.csv.gz" + manifest_path = output_dir / "manifest.json" + lifecycle_panel.to_csv(panel_path, index=False, compression="gzip") + market_history.to_csv(market_path, index=False, compression="gzip") + manifest = { + "contract_version": "crypto.lifecycle_preflight.v1", + "panel_rows": int(len(lifecycle_panel)), + "panel_symbols": sorted(lifecycle_panel["symbol"].unique().tolist()), + "market_rows": int(len(market_history)), + "market_symbols": sorted(market_history["symbol"].unique().tolist()), + "start_date": panel_dates.min().date().isoformat(), + "end_date": panel_dates.max().date().isoformat(), + } + manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return manifest + + +def main() -> int: + parser = argparse.ArgumentParser(description="Export real production research inputs for lifecycle drift preflight.") + parser.add_argument("--config", default="config/default.yaml") + parser.add_argument("--universe-mode", default="broad_liquid") + parser.add_argument("--output-dir", type=Path, required=True) + args = parser.parse_args() + + config = load_config(args.config) + result = run_research_pipeline(config, universe_mode=args.universe_mode) + manifest = export_lifecycle_inputs(result["panel"], args.output_dir) + print(manifest) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_export_lifecycle_preflight_inputs.py b/tests/test_export_lifecycle_preflight_inputs.py new file mode 100644 index 0000000..d6180a5 --- /dev/null +++ b/tests/test_export_lifecycle_preflight_inputs.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path + +import pandas as pd + +from scripts.export_lifecycle_preflight_inputs import export_lifecycle_inputs + + +class ExportLifecyclePreflightInputsTests(unittest.TestCase): + @staticmethod + def _valid_panel() -> pd.DataFrame: + dates = pd.date_range(end=pd.Timestamp.now().normalize(), periods=1000, freq="D") + symbols = ("BTCUSDT", "ETHUSDT", "SOLUSDT") + index = pd.MultiIndex.from_product([dates, symbols], names=["date", "symbol"]) + panel = pd.DataFrame(index=index) + panel["in_universe"] = True + panel["open"] = range(1, len(panel) + 1) + panel["close"] = panel["open"] + 0.5 + panel["final_score"] = 0.75 + return panel + + def test_export_lifecycle_inputs_writes_real_panel_contract(self) -> None: + panel = self._valid_panel() + with tempfile.TemporaryDirectory() as tmpdir: + output_dir = Path(tmpdir) + manifest = export_lifecycle_inputs(panel, output_dir) + + exported_panel = pd.read_csv(output_dir / "research_panel.csv.gz") + market_history = pd.read_csv(output_dir / "market_history.csv.gz") + persisted_manifest = json.loads((output_dir / "manifest.json").read_text()) + + self.assertEqual( + set(exported_panel.columns), + {"date", "symbol", "in_universe", "open", "final_score"}, + ) + self.assertEqual(set(market_history["symbol"]), {"BTCUSDT", "ETHUSDT"}) + self.assertEqual(manifest, persisted_manifest) + self.assertEqual(manifest["contract_version"], "crypto.lifecycle_preflight.v1") + + def test_export_rejects_scored_date_without_universe(self) -> None: + panel = self._valid_panel() + latest_completed_date = panel.index.get_level_values("date").unique()[-2] + panel.loc[(latest_completed_date, slice(None)), "in_universe"] = False + + with tempfile.TemporaryDirectory() as tmpdir, self.assertRaisesRegex( + ValueError, + "at least two in-universe symbols per scored date", + ): + export_lifecycle_inputs(panel, Path(tmpdir)) + + def test_export_preserves_open_rows_without_scores(self) -> None: + panel = self._valid_panel() + date = panel.index.get_level_values("date").unique()[500] + panel.loc[(date, "BTCUSDT"), ["in_universe", "final_score"]] = [False, pd.NA] + + with tempfile.TemporaryDirectory() as tmpdir: + output_dir = Path(tmpdir) + export_lifecycle_inputs(panel, output_dir) + exported = pd.read_csv(output_dir / "research_panel.csv.gz") + + row = exported.loc[ + (exported["date"] == date.date().isoformat()) + & (exported["symbol"] == "BTCUSDT") + ] + self.assertEqual(len(row), 1) + self.assertTrue(pd.isna(row.iloc[0]["final_score"])) + + def test_export_rejects_stale_combo_history_even_when_panel_is_fresh(self) -> None: + panel = self._valid_panel() + cutoff = panel.index.get_level_values("date").max() - pd.Timedelta(days=10) + combo_mask = panel.index.get_level_values("symbol").isin({"BTCUSDT", "ETHUSDT"}) + stale_mask = combo_mask & (panel.index.get_level_values("date") > cutoff) + panel.loc[stale_mask, "close"] = pd.NA + + with tempfile.TemporaryDirectory() as tmpdir, self.assertRaisesRegex( + ValueError, + "BTC/ETH market history is stale", + ): + export_lifecycle_inputs(panel, Path(tmpdir)) diff --git a/tests/test_publish_lifecycle_inputs_workflow.py b/tests/test_publish_lifecycle_inputs_workflow.py new file mode 100644 index 0000000..a039b74 --- /dev/null +++ b/tests/test_publish_lifecycle_inputs_workflow.py @@ -0,0 +1,16 @@ +from pathlib import Path + + +def test_publish_lifecycle_inputs_workflow_uses_real_research_pipeline() -> None: + workflow = ( + Path(__file__).resolve().parents[1] / ".github" / "workflows" / "publish-lifecycle-inputs.yml" + ).read_text(encoding="utf-8") + + assert 'DOWNLOAD_TOP_LIQUID: "90"' in workflow + assert 'scripts/download_history.py --top-liquid "${DOWNLOAD_TOP_LIQUID}"' in workflow + assert '--end-date "${completed_date}"' in workflow + assert 'scripts/download_history.py --symbols ETHUSDT' in workflow + assert "scripts/export_lifecycle_preflight_inputs.py" in workflow + assert "--universe-mode broad_liquid" in workflow + assert "crypto-lifecycle-inputs-${{ github.run_id }}-${{ github.run_attempt }}" in workflow + assert "if-no-files-found: error" in workflow