diff --git a/rdagent/components/coder/kronos_adapter.py b/rdagent/components/coder/kronos_adapter.py new file mode 100644 index 00000000..b7b15d55 --- /dev/null +++ b/rdagent/components/coder/kronos_adapter.py @@ -0,0 +1,274 @@ +""" +Kronos Foundation Model Adapter for Predix. + +Wraps the Kronos-mini OHLCV foundation model (4.1M params, AAAI 2026, MIT) +for use as: + - Factor (Option A): predicted next-day return signal + - Model alongside LightGBM (Option B): IC/Sharpe evaluation + +Kronos repo: https://github.com/shiyu-coder/Kronos +HuggingFace: NeoQuasar/Kronos-mini | NeoQuasar/Kronos-Tokenizer-2k +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Optional + +import numpy as np +import pandas as pd +import torch + +from rdagent.log import rdagent_logger as logger + +KRONOS_REPO = Path.home() / "Kronos" +_KRONOS_AVAILABLE: Optional[bool] = None + + +def _ensure_kronos() -> bool: + global _KRONOS_AVAILABLE + if _KRONOS_AVAILABLE is not None: + return _KRONOS_AVAILABLE + if not KRONOS_REPO.exists(): + logger.warning(f"Kronos repo not found at {KRONOS_REPO}. Clone with: git clone https://github.com/shiyu-coder/Kronos ~/Kronos") + _KRONOS_AVAILABLE = False + return False + repo_str = str(KRONOS_REPO) + if repo_str not in sys.path: + sys.path.insert(0, repo_str) + try: + import model as _ # noqa: F401 + _KRONOS_AVAILABLE = True + except ImportError as e: + logger.warning(f"Failed to import Kronos model: {e}") + _KRONOS_AVAILABLE = False + return _KRONOS_AVAILABLE + + +def _ohlcv_from_predix(df: pd.DataFrame) -> pd.DataFrame: + """Convert Predix HDF5 format ($open/$close/...) to Kronos format (open/close/...).""" + col_map = {"$open": "open", "$high": "high", "$low": "low", "$close": "close", "$volume": "volume"} + renamed = df.rename(columns=col_map) + cols = [c for c in ["open", "high", "low", "close", "volume"] if c in renamed.columns] + return renamed[cols].astype(float) + + +class KronosAdapter: + """ + Loads Kronos-mini once and provides rolling-window OHLCV inference. + + Usage: + adapter = KronosAdapter(device="cuda") + adapter.load() + pred_return = adapter.predict_return(ohlcv_df, context_bars=512, pred_bars=96) + """ + + MODEL_ID = "NeoQuasar/Kronos-mini" + TOKENIZER_ID = "NeoQuasar/Kronos-Tokenizer-2k" + + def __init__(self, device: str = "cuda" if torch.cuda.is_available() else "cpu", max_context: int = 512): + self.device = device + self.max_context = max_context + self._predictor = None + + def load(self) -> "KronosAdapter": + if self._predictor is not None: + return self + if not _ensure_kronos(): + raise RuntimeError("Kronos not available — see warning above.") + from model import Kronos, KronosTokenizer, KronosPredictor # type: ignore + + logger.info(f"Loading Kronos-mini from HuggingFace ({self.MODEL_ID})...") + tokenizer = KronosTokenizer.from_pretrained(self.TOKENIZER_ID) + model = Kronos.from_pretrained(self.MODEL_ID) + self._predictor = KronosPredictor(model, tokenizer, device=self.device, max_context=self.max_context) + logger.info("Kronos-mini loaded.") + return self + + def predict_next_bars( + self, + ohlcv_df: pd.DataFrame, + context_bars: int, + pred_bars: int, + temperature: float = 1.0, + top_p: float = 0.9, + ) -> pd.DataFrame: + """ + Run Kronos on `context_bars` of OHLCV data, returning `pred_bars` predicted bars. + + Args: + ohlcv_df: DataFrame with columns open/high/low/close[/volume], DatetimeIndex. + context_bars: Number of history bars to feed as context. + pred_bars: Number of future bars to predict. + + Returns: + DataFrame with predicted open/high/low/close/volume, indexed by future timestamps. + """ + if self._predictor is None: + raise RuntimeError("Call .load() first.") + if len(ohlcv_df) < context_bars: + raise ValueError(f"Need at least {context_bars} bars, got {len(ohlcv_df)}") + + ctx = ohlcv_df.iloc[-context_bars:].copy().reset_index(drop=True) + freq = ohlcv_df.index.freq or pd.infer_freq(ohlcv_df.index[:100]) + last_ts = ohlcv_df.index[-1] + future_idx = pd.date_range(start=last_ts, periods=pred_bars + 1, freq=freq or "1min")[1:] + + x_timestamp = pd.Series(ctx.index) + y_timestamp = pd.Series(range(len(ctx), len(ctx) + pred_bars)) + + pred_df = self._predictor.predict( + df=ctx, + x_timestamp=x_timestamp, + y_timestamp=y_timestamp, + pred_len=pred_bars, + T=temperature, + top_p=top_p, + sample_count=1, + verbose=False, + ) + pred_df.index = future_idx + return pred_df + + def predict_return( + self, + ohlcv_df: pd.DataFrame, + context_bars: int = 512, + pred_bars: int = 1, + ) -> float: + """ + Predict the average return over the next `pred_bars` using the last `context_bars`. + Returns the predicted log-return (predicted_close / last_close - 1). + """ + pred = self.predict_next_bars(ohlcv_df, context_bars=context_bars, pred_bars=pred_bars) + last_close = float(ohlcv_df["close"].iloc[-1]) + pred_close = float(pred["close"].iloc[-1]) + return pred_close / last_close - 1.0 + + +def build_kronos_factor( + hdf5_path: str | Path, + context_bars: int = 512, + pred_bars: int = 96, + stride_bars: int = 96, + device: str = "cuda" if torch.cuda.is_available() else "cpu", +) -> pd.DataFrame: + """ + Generate the Kronos predicted-return factor for all EUR/USD 1-min bars. + + Strategy: + Every `stride_bars` bars, run Kronos on the previous `context_bars` and + predict the next `pred_bars`. The predicted log-return is forward-filled + across the predicted window. Default: daily stride (96 bars/day at 1-min). + + Returns: + MultiIndex (datetime, instrument) DataFrame with column "KronosPredReturn". + """ + logger.info(f"Loading data from {hdf5_path}...") + raw = pd.read_hdf(hdf5_path, key="data") + + # Flatten to single-instrument 1-min series + instrument = raw.index.get_level_values("instrument").unique()[0] + df = raw.xs(instrument, level="instrument") + ohlcv = _ohlcv_from_predix(df) + + adapter = KronosAdapter(device=device, max_context=min(context_bars, 512)) + adapter.load() + + factor_values: dict[pd.Timestamp, float] = {} + n = len(ohlcv) + starts = range(context_bars, n, stride_bars) + + logger.info(f"Running Kronos inference: {len(starts)} windows (stride={stride_bars}, ctx={context_bars}, pred={pred_bars})") + for i, bar_idx in enumerate(starts): + ctx_df = ohlcv.iloc[bar_idx - context_bars:bar_idx] + try: + pred = adapter.predict_next_bars(ctx_df, context_bars=context_bars, pred_bars=pred_bars) + last_close = float(ctx_df["close"].iloc[-1]) + # Store predicted return for each predicted bar's timestamp + for ts, row in pred.iterrows(): + ret = float(row["close"]) / last_close - 1.0 + factor_values[ts] = ret + except Exception as e: + logger.warning(f"Kronos inference failed at bar {bar_idx}: {e}") + continue + + if (i + 1) % 100 == 0: + logger.info(f" {i+1}/{len(starts)} windows done") + + if not factor_values: + raise RuntimeError("No Kronos predictions were generated.") + + factor_series = pd.Series(factor_values, name="KronosPredReturn") + factor_series = factor_series.reindex(ohlcv.index, method="ffill") + + result = factor_series.to_frame() + result.index = pd.MultiIndex.from_arrays( + [ohlcv.index, [instrument] * len(ohlcv)], + names=["datetime", "instrument"], + ) + logger.info(f"Kronos factor built: {len(result)} bars, {result['KronosPredReturn'].notna().sum()} non-NaN") + return result + + +def evaluate_kronos_model( + hdf5_path: str | Path, + context_bars: int = 512, + pred_bars: int = 30, + stride_bars: int = 30, + device: str = "cuda" if torch.cuda.is_available() else "cpu", +) -> dict: + """ + Evaluate Kronos as a standalone model (Option B, alongside LightGBM). + + Computes IC (Information Coefficient) between Kronos predicted returns and + actual realized returns on the test set. + + Returns: + dict with keys: IC_mean, IC_std, IC_IR (IC / std), hit_rate, n_predictions + """ + raw = pd.read_hdf(hdf5_path, key="data") + instrument = raw.index.get_level_values("instrument").unique()[0] + df = raw.xs(instrument, level="instrument") + ohlcv = _ohlcv_from_predix(df) + + adapter = KronosAdapter(device=device, max_context=min(context_bars, 512)) + adapter.load() + + predicted_returns = [] + actual_returns = [] + n = len(ohlcv) + + for bar_idx in range(context_bars, n - pred_bars, stride_bars): + ctx_df = ohlcv.iloc[bar_idx - context_bars:bar_idx] + try: + pred = adapter.predict_next_bars(ctx_df, context_bars=context_bars, pred_bars=pred_bars) + last_close = float(ctx_df["close"].iloc[-1]) + pred_ret = float(pred["close"].iloc[-1]) / last_close - 1.0 + + actual_future = ohlcv.iloc[bar_idx:bar_idx + pred_bars] + actual_ret = float(actual_future["close"].iloc[-1]) / last_close - 1.0 + + predicted_returns.append(pred_ret) + actual_returns.append(actual_ret) + except Exception: + continue + + pred_arr = np.array(predicted_returns) + actual_arr = np.array(actual_returns) + + ic = np.corrcoef(pred_arr, actual_arr)[0, 1] if len(pred_arr) > 1 else float("nan") + ic_std = float(np.std([ + np.corrcoef(pred_arr[i:i+50], actual_arr[i:i+50])[0, 1] + for i in range(0, len(pred_arr) - 50, 10) + ])) if len(pred_arr) > 60 else float("nan") + hit_rate = float(np.mean(np.sign(pred_arr) == np.sign(actual_arr))) + + return { + "IC_mean": float(ic), + "IC_std": ic_std, + "IC_IR": float(ic / ic_std) if ic_std and ic_std > 0 else float("nan"), + "hit_rate": hit_rate, + "n_predictions": len(pred_arr), + } diff --git a/scripts/kronos_factor_gen.py b/scripts/kronos_factor_gen.py new file mode 100644 index 00000000..c18cc68e --- /dev/null +++ b/scripts/kronos_factor_gen.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python +""" +Option A: Generate Kronos predicted-return factor from EUR/USD 1-min data. + +Runs Kronos-mini inference in daily strides (96 bars/day) over all available +OHLCV data and saves the resulting factor for use in Predix's factor pipeline. + +Usage: + conda activate predix + python scripts/kronos_factor_gen.py + python scripts/kronos_factor_gen.py --context 512 --pred 96 --device cuda + python scripts/kronos_factor_gen.py --device cpu # slower but no GPU needed +""" + +import argparse +import json +from datetime import datetime +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv(Path(__file__).parent.parent / ".env") + +import pandas as pd +import torch + +DATA_PATH = Path("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") +OUTPUT_DIR = Path("results/factors") + + +def main(): + parser = argparse.ArgumentParser(description="Generate Kronos predicted-return factor") + parser.add_argument("--context", type=int, default=512, help="Context window in bars (max 512 for Kronos-mini)") + parser.add_argument("--pred", type=int, default=96, help="Prediction horizon in bars (default: 96 = 1 trading day)") + parser.add_argument("--stride", type=int, default=None, help="Stride between windows (default: same as --pred)") + parser.add_argument("--device", type=str, default="cuda" if torch.cuda.is_available() else "cpu") + parser.add_argument("--output", type=str, default=None, help="Output parquet path (default: auto)") + args = parser.parse_args() + + stride = args.stride or args.pred + + print(f"Kronos Factor Generator") + print(f" Data: {DATA_PATH}") + print(f" Context: {args.context} bars") + print(f" Pred: {args.pred} bars ({args.pred} min = {args.pred/96:.1f} trading days)") + print(f" Stride: {stride} bars") + print(f" Device: {args.device}") + print() + + if not DATA_PATH.exists(): + print(f"ERROR: Data not found at {DATA_PATH}") + print("Run data conversion first — see README Data Setup section.") + raise SystemExit(1) + + from rdagent.components.coder.kronos_adapter import build_kronos_factor + + factor_df = build_kronos_factor( + hdf5_path=DATA_PATH, + context_bars=args.context, + pred_bars=args.pred, + stride_bars=stride, + device=args.device, + ) + + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + out_path = args.output or OUTPUT_DIR / f"kronos_pred_return_p{args.pred}.parquet" + factor_df.to_parquet(out_path) + print(f"\nFactor saved to: {out_path}") + print(f"Shape: {factor_df.shape}") + print(f"Non-NaN: {factor_df['KronosPredReturn'].notna().sum()}") + print(f"\nSample (first 5):") + print(factor_df.head()) + + # Save metadata for predix.py top / best integration + meta = { + "factor_name": f"KronosPredReturn_p{args.pred}", + "description": f"Kronos-mini predicted return, {args.pred}-bar horizon", + "model": "NeoQuasar/Kronos-mini", + "context_bars": args.context, + "pred_bars": args.pred, + "stride_bars": stride, + "device": args.device, + "generated_at": datetime.now().isoformat(), + "n_bars": len(factor_df), + "n_non_nan": int(factor_df["KronosPredReturn"].notna().sum()), + "parquet_path": str(out_path), + } + meta_path = out_path.with_suffix(".json") + with open(meta_path, "w") as f: + json.dump(meta, f, indent=2) + print(f"Metadata saved to: {meta_path}") + + +if __name__ == "__main__": + main() diff --git a/scripts/kronos_model_eval.py b/scripts/kronos_model_eval.py new file mode 100644 index 00000000..aad65123 --- /dev/null +++ b/scripts/kronos_model_eval.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python +""" +Option B: Evaluate Kronos-mini as a model alongside LightGBM. + +Computes IC (Information Coefficient) and hit rate for Kronos predictions +vs actual realized returns. Results are printed for comparison with LightGBM. + +Usage: + conda activate predix + python scripts/kronos_model_eval.py + python scripts/kronos_model_eval.py --pred 30 --context 512 --device cuda +""" + +import argparse +import json +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv(Path(__file__).parent.parent / ".env") + +import torch + +DATA_PATH = Path("git_ignore_folder/factor_implementation_source_data/intraday_pv.h5") +OUTPUT_DIR = Path("results/kronos") + + +def main(): + parser = argparse.ArgumentParser(description="Evaluate Kronos as model (alongside LightGBM)") + parser.add_argument("--context", type=int, default=512, help="Context window in bars") + parser.add_argument("--pred", type=int, default=30, help="Prediction horizon in bars") + parser.add_argument("--stride", type=int, default=None, help="Stride between evaluations (default: pred)") + parser.add_argument("--device", type=str, default="cuda" if torch.cuda.is_available() else "cpu") + args = parser.parse_args() + + stride = args.stride or args.pred + + print(f"Kronos Model Evaluator (alongside LightGBM)") + print(f" Context: {args.context} bars | Pred: {args.pred} bars | Device: {args.device}") + print() + + if not DATA_PATH.exists(): + print(f"ERROR: Data not found at {DATA_PATH}") + raise SystemExit(1) + + from rdagent.components.coder.kronos_adapter import evaluate_kronos_model + + print("Running evaluation (this may take several minutes)...") + metrics = evaluate_kronos_model( + hdf5_path=DATA_PATH, + context_bars=args.context, + pred_bars=args.pred, + stride_bars=stride, + device=args.device, + ) + + print("\n" + "=" * 50) + print("Kronos-mini Model Evaluation Results") + print("=" * 50) + print(f" Predictions: {metrics['n_predictions']}") + print(f" IC (mean): {metrics['IC_mean']:.4f}") + print(f" IC (std): {metrics['IC_std']:.4f}") + print(f" IC IR: {metrics['IC_IR']:.4f} (>0.5 = good)") + print(f" Hit Rate: {metrics['hit_rate']:.2%} (>50% = directionally useful)") + print("=" * 50) + print() + print("Reference: LightGBM baseline IC typically 0.01–0.05 on 1-min EUR/USD") + + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + out = OUTPUT_DIR / f"kronos_eval_ctx{args.context}_pred{args.pred}.json" + with open(out, "w") as f: + json.dump({**metrics, "context_bars": args.context, "pred_bars": args.pred}, f, indent=2) + print(f"\nResults saved to: {out}") + + +if __name__ == "__main__": + main() diff --git a/test/backtesting/test_kronos_adapter.py b/test/backtesting/test_kronos_adapter.py new file mode 100644 index 00000000..96958146 --- /dev/null +++ b/test/backtesting/test_kronos_adapter.py @@ -0,0 +1,102 @@ +"""Tests for KronosAdapter — mock-based, no real model download needed.""" + +import numpy as np +import pandas as pd +import pytest + + +def _make_ohlcv(n: int = 600) -> pd.DataFrame: + """Synthetic 1-min OHLCV DataFrame.""" + idx = pd.date_range("2024-01-01", periods=n, freq="1min") + close = 1.1000 + np.cumsum(np.random.randn(n) * 0.0001) + df = pd.DataFrame({ + "open": close + np.random.randn(n) * 0.00005, + "high": close + np.abs(np.random.randn(n) * 0.0001), + "low": close - np.abs(np.random.randn(n) * 0.0001), + "close": close, + "volume": np.abs(np.random.randn(n) * 100), + }, index=idx) + return df + + +def test_ohlcv_conversion(): + """_ohlcv_from_predix renames $ columns correctly.""" + from rdagent.components.coder.kronos_adapter import _ohlcv_from_predix + + idx = pd.MultiIndex.from_arrays( + [pd.date_range("2024-01-01", periods=3, freq="1min"), ["EURUSD"] * 3], + names=["datetime", "instrument"], + ) + predix_df = pd.DataFrame({ + "$open": [1.1, 1.2, 1.3], + "$high": [1.15, 1.25, 1.35], + "$low": [1.05, 1.15, 1.25], + "$close": [1.12, 1.22, 1.32], + "$volume": [100.0, 200.0, 300.0], + }, index=idx) + + ohlcv = _ohlcv_from_predix(predix_df) + assert "close" in ohlcv.columns + assert "$close" not in ohlcv.columns + assert list(ohlcv.columns) == ["open", "high", "low", "close", "volume"] + + +def test_kronos_adapter_load_skipped_without_repo(tmp_path, monkeypatch): + """KronosAdapter gracefully reports unavailable when repo is missing.""" + import rdagent.components.coder.kronos_adapter as mod + monkeypatch.setattr(mod, "KRONOS_REPO", tmp_path / "nonexistent") + monkeypatch.setattr(mod, "_KRONOS_AVAILABLE", None) + + from rdagent.components.coder.kronos_adapter import KronosAdapter, _ensure_kronos + available = _ensure_kronos() + assert available is False + + +def test_build_kronos_factor_mock(tmp_path, monkeypatch): + """build_kronos_factor produces correct MultiIndex output with mocked predictor.""" + import rdagent.components.coder.kronos_adapter as mod + + # Mock the adapter so no real Kronos load happens + class MockAdapter: + def load(self): return self + def predict_next_bars(self, ohlcv_df, context_bars, pred_bars, **kw): + idx = pd.date_range(ohlcv_df.index[-1], periods=pred_bars + 1, freq="1min")[1:] + last_close = float(ohlcv_df["close"].iloc[-1]) + return pd.DataFrame({ + "open": last_close * (1 + np.random.randn(pred_bars) * 0.001), + "close": last_close * (1 + np.random.randn(pred_bars) * 0.001), + "high": last_close * 1.001, + "low": last_close * 0.999, + "volume": 100.0, + }, index=idx) + + monkeypatch.setattr(mod, "KronosAdapter", lambda **kw: MockAdapter()) + + # Write minimal HDF5 + n = 300 + idx = pd.MultiIndex.from_arrays( + [pd.date_range("2024-01-01", periods=n, freq="1min"), ["EURUSD"] * n], + names=["datetime", "instrument"], + ) + df = pd.DataFrame({ + "$open": np.random.rand(n).astype("float32") + 1.1, + "$close": np.random.rand(n).astype("float32") + 1.1, + "$high": np.random.rand(n).astype("float32") + 1.11, + "$low": np.random.rand(n).astype("float32") + 1.09, + "$volume": np.random.rand(n).astype("float32") * 100, + }, index=idx) + h5_path = tmp_path / "intraday_pv.h5" + df.to_hdf(h5_path, key="data", mode="w") + + result = mod.build_kronos_factor( + hdf5_path=h5_path, + context_bars=100, + pred_bars=20, + stride_bars=20, + device="cpu", + ) + + assert isinstance(result, pd.DataFrame) + assert result.index.names == ["datetime", "instrument"] + assert "KronosPredReturn" in result.columns + assert result["KronosPredReturn"].notna().sum() > 0