kronos-finance

Pythonic wrapper around the Kronos foundation model for OHLCV forecasting across any market.

Built on Kronos

Kronos is the first open-source foundation model for financial candlesticks (K-lines), by the NeoQuasar team, accepted at AAAI 2026, MIT-licensed.

This package (kronos-finance) is a wrapper that turns the upstream research codebase into a pip-installable library with a CLI, a dashboard, multi-source data loaders, and comprehensive tests. All model code comes from the original project — see the Citation section.

• Original repo: https://github.com/shiyu-coder/Kronos
• Paper: https://arxiv.org/abs/2508.02739
• Model zoo on HuggingFace: https://huggingface.co/NeoQuasar

Use a virtual environment (recommended)

A virtual environment keeps your system Python clean and avoids the error: externally-managed-environment (PEP 668) error on Ubuntu 23.04+, macOS Homebrew Python, and Fedora 39+.

python -m venv .venv
source .venv/bin/activate          # macOS / Linux
# .venv\Scripts\activate           # Windows PowerShell

Every install command below assumes you've activated a venv first.

Installation

pip install kronos-finance                       # core (CUDA-enabled PyTorch, ~800MB)
pip install kronos-finance[cn]                   # + AKShare for Chinese A-shares
pip install kronos-finance[global]               # + yfinance for global equities
pip install kronos-finance[crypto]               # + CCXT for crypto exchanges
pip install kronos-finance[qlib]                 # + Qlib for CN finetune data
pip install kronos-finance[ui]                   # + Flask + Plotly for the dashboard
pip install kronos-finance[analysis]             # + pandas-ta + quantstats
pip install kronos-finance[all]                  # everything above

The default install includes CUDA-enabled PyTorch and works on CPU or GPU. For CPU-only or a specific CUDA version, see Hardware below before installing.

Verify the install:

kronos --version

Quickstart (60 seconds)

from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv
import pandas as pd

df = load_ohlcv("AAPL", period="2y", interval="1d")
wrapper = load_kronos(model_id="small")
y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
pred = wrapper.predict(
    df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
    x_timestamp=df["timestamps"].tail(400), y_timestamp=y_ts, pred_len=30,
)
print(pred.head())

Features

• One-line predict on any market — US equities, CN A-shares, crypto, HK, JP, EU.
• Multi-source loaders — AKShare, yfinance, CCXT, Qlib, local CSV.
• Ticker auto-detection — type "600519" and AKShare is picked; type "BTC/USDT" and CCXT is picked.
• Bundled ticker catalog + user-extendable ~/.kronos/tickers.json.
• CLI: kronos predict, kronos backtest, kronos ui, kronos tickers.
• Flask dashboard with candlestick chart, autocomplete, model selector, indicator overlay.
• Optional indicators (RSI, MACD, BBands) and HTML tearsheets (quantstats).
• 12 runnable examples + comprehensive docs + a 35-term glossary.

How to predict any ticker in the world

Market	Ticker format	Source	Install extra
US equity	AAPL, MSFT, NVDA	yfinance	[global]
US ETF	SPY, QQQ, IWM	yfinance	[global]
US index	^GSPC, ^DJI, ^IXIC	yfinance	[global]
CN A-share	600519, 000001, 002594	AKShare	[cn]
CN index	000300, 000905	AKShare	[cn]
Crypto pair	BTC/USDT, ETH/USDT	CCXT (Binance default)	[crypto]
HK stock	0700.HK, 9988.HK	yfinance	[global]
JP stock	7203.T, 6758.T	yfinance	[global]
EU stock	ASML.AS, SAP.DE	yfinance	[global]
Local CSV	path to .csv	csv_path=	(core)

The source="auto" default routes the ticker to the right loader based on its shape.

CLI reference

kronos predict TICKER [--source auto] [--model small] [--interval 1d]
                     [--pred-len 30] [--lookback 400] [--export forecast.csv]
                     [--device cpu]
kronos batch TICKERS_FILE [--output ./out] [--model small] [--pred-len 30]
kronos backtest TICKER [--period 1y] [--interval 1d] [--source auto]
                      [--export tearsheet.html]
kronos ui [--host 127.0.0.1] [--port 5000] [--debug]
kronos tickers list
kronos tickers search QUERY [--market cn|us|crypto|...]
kronos tickers add SYMBOL NAME SOURCE [--exchange binance]
kronos --version
kronos --help

Examples:

kronos predict 600519 --source akshare
kronos predict BTC/USDT --source ccxt --exchange binance

Exit codes: 0 ok, 1 generic error, 2 usage error, 130 SIGINT (clean Ctrl+C).

Python API

from kronos_finance import load_kronos          # model
from kronos_finance.data import load_ohlcv       # data fetchers
from kronos_finance.tickers import catalog, search, add_user_ticker
from kronos_finance.analysis import (
    enrich_with_indicators, forecast_to_returns, make_tearsheet,
)
from kronos_finance import (                      # errors
    KronosFinanceError, TickerNotFoundError,
    DataSourceError, ModelLoadError, PredictionError, CatalogError,
)

Full reference: docs/API.md.

Dashboard

Launch with kronos ui (default http://127.0.0.1:5000):

• Ticker input with autocomplete from the bundled + user catalog.
• Candlestick chart with historical in green/red and forecast in blue/purple.
• Model selector (mini / small / base with parameter counts).
• Source + interval pickers.
• Indicator overlay (RSI / MACD / BBands).
• Recent predictions history with CSV export.
• Dark / light theme toggle.

Dashboard quickstart

1. kronos ui from your shell.
2. Open http://localhost:5000 in any modern browser.
3. Type a ticker (e.g. AAPL, 600519, BTC/USDT).
4. Click Predict.
5. The forecast paints on the chart; the metrics card shows the predicted close and expected return; the row appears in the Recent predictions table.
6. Click CSV on any row to download that prediction.

The dashboard is also a JSON API. You can drive it from any HTTP client — see examples/22_dashboard_api_reference.py for curl, Python requests, and JavaScript fetch examples for every endpoint.

Full tour (anatomy of the page, every UI element, every error message, production deployment, extending the dashboard): docs/DASHBOARD.md.

Examples — full runnable source

Important — read first. The source repo (https://github.com/lordxmen2k/kronos-finance) is private, so the examples/ directory is not visible to PyPI users. The complete runnable Python and shell source for every example is reproduced below in full — copy any block into a file (e.g., my_example.py) and run it. Each example carries a top-of-file docstring with WHAT THIS DEMONSTRATES / WHEN TO USE THIS / EXPECTED OUTPUT / COMMON PITFALLS / BEFORE RUNNING / USAGE so you know exactly what you're getting.

Index

#	File	Skill	Theme	Extra
01	01_quickstart_predict.py	Beginner	Single-ticker 30-day forecast	[global]
02	02_us_stock_yfinance.py	Beginner	Multi-ticker US watchlist with overlay chart	[global]
03	03_cn_a_share_akshare.py	Beginner	CN A-share with column-rename walkthrough	[cn]
04	04_crypto_ccxt.py	Beginner	BTC/USDT 5-minute bars from Binance	[crypto]
05	05_batch_predict.py	Intermediate	Batch sweep across many tickers	[global]
06	06_indicators_and_tearsheet.py	Intermediate	RSI/MACD/Bollinger + quantstats tearsheet	[global,analysis]
07	07_cli_predict.sh	Beginner	kronos predict from a shell script	[global]
08	08_cli_backtest.sh	Intermediate	kronos backtest from a shell script	[global]
09	09_launch_dashboard.sh	Beginner	Start the Flask dashboard	[ui]
10	10_save_load_predictions.py	Intermediate	CSV / JSON / Parquet save/load round-trip	[global,analysis]
11	11_qlib_finetune_smoke.py	Advanced	Qlib-format CSV for the upstream finetune script	[qlib]
12	12_custom_model_local.py	Advanced	Load a local checkpoint instead of HuggingFace	(none)
13	13_multi_timeframe.py	Advanced	Daily + hourly + 5-min forecasts on one ticker	[global]
14	14_quantile_bands.py	Advanced	P10/P25/P50/P75/P90 fan chart from sample paths	[global]
15	15_recursive_predict.py	Advanced	1-year forecast via autoregressive chunking	[global]
16	16_csv_with_arbitrary_columns.py	Intermediate	Adapt any CSV (English / Chinese / custom)	(none)
17	17_healthcheck_environment.py	Intermediate	CI-friendly env health check (JSON output opt-in)	(none)
18	18_failure_modes_and_recovery.py	Reference	Runbook of 6 common failures + recovery	(none)
19	19_cronjob_daily_forecast.py	Advanced	Idempotent daily job with lockfile + logs	[global]
20	20_local_csv_user_data.py	Intermediate	Forecast entirely from a local CSV (no network)	(none)
21	21_portfolio_kronos_weighting.py	Advanced	Three weighting schemes for portfolio construction	[global]
22	22_dashboard_api_reference.py	Reference	curl / Python / JS examples for the dashboard JSON API	(none)

Quick-pick by goal:

• "Show me how to forecast one ticker" → Example 01
• "Run a daily forecast across my watchlist" → Example 19
• "Evaluate Kronos vs buy-and-hold" → Example 06
• "I have my own CSV / proprietary data" → Example 16 or 20
• "Troubleshoot my installation" → Example 17 then 18
• "Build a portfolio with Kronos signals" → Example 21
• "Wire Kronos into a broker / live trading" → Trading Strategy use cases section below

Example 01 — 01_quickstart_predict.py

Skill: Beginner. Theme: Single-ticker 30-day forecast. Extra needed: [global].

Run: python 01_quickstart_predict.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 01 — Quickstart: predict AAPL's next 30 days.

WHAT THIS DEMONSTRATES
-----------------------
The smallest possible end-to-end forecast:
  1. Fetch 2 years of AAPL daily bars from yfinance.
  2. Load the Kronos-small model (24.7M params, ~95 MB download).
  3. Build the future-timestamp index (one bar per day, 30 days).
  4. Call predict() with explicit sampling parameters.
  5. Print the forecasted OHLCV table.
  6. Plot the close-price trajectory alongside the historical context.

WHEN TO USE THIS
----------------
You have a CSV (or a ticker), you want to know "what does Kronos think
happens next?" and you want to see the result immediately. This is the
quickest path from zero to a chart.

EXPECTED OUTPUT
---------------
- A DataFrame table with 30 rows (one per predicted business day).
- A PNG file `aapl_forecast.png` showing historical close (last 120
  bars) plus the 30-day forecast trace.
- First run downloads ~95 MB of model weights; later runs are instant.

COMMON PITFALLS
---------------
- **PEP 668 "externally-managed-environment"**: you forgot the venv. Create
  one and activate it before pip install.
- **Model download stalls**: first run needs internet to fetch weights from
  HuggingFace. Set HF_HOME=/path/with/space if your home dir is small.
- **"No module named 'yfinance'"**: install with `pip install
  kronos-finance[global]`, not just `pip install kronos-finance`.
- **Empty yfinance response**: ticker may be delisted; try a different
  period (e.g. `period="5y"` instead of `period="2y"`).
- **CUDA version mismatch** if you ran pip install torch without setting
  the right index URL; reinstall torch from the matching PyTorch index.

BEFORE RUNNING
--------------
    python -m venv .venv
    source .venv/bin/activate        # or .venv\\Scripts\\activate on Windows
    pip install kronos-finance[global]

USAGE
-----
    python examples/01_quickstart_predict.py
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib.pyplot as plt
import pandas as pd

# Add the src dir to sys.path if running this example from a fresh checkout
# without first doing `pip install -e .`. Comment out if you've installed.
_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def fetch_history(ticker: str, period: str = "2y") -> pd.DataFrame:
    """Download N years of daily OHLCV bars for the ticker.

    Returns a standardized DataFrame:
        columns = timestamps, open, high, low, close, volume, amount
        tz-naive timestamps, ascending order.
    """
    df = load_ohlcv(ticker, period=period, interval="1d")
    if len(df) < 400:
        # The Kronos-small context length is 512 bars; we need at least
        # 30 bars of headroom beyond the model's preferred 400-bar lookback.
        print(f"warning: only {len(df)} bars fetched; results may be poor.")
    return df


def build_future_index(last_ts: pd.Timestamp, periods: int, freq: str = "1D") -> pd.DatetimeIndex:
    """Build the prediction-time index.

    We start at last_ts + 1 freq and generate `periods` timestamps.
    Note: pd.date_range is inclusive on both ends, so we slice off [0].
    """
    return pd.date_range(start=last_ts, periods=periods + 1, freq=freq)[1:]


def main() -> int:
    ticker = "AAPL"
    print(f"=== kronos-finance quickstart: {ticker} ===\n")

    # 1. Fetch history
    print(f"Step 1/4: fetching 2 years of {ticker} daily bars...")
    df = fetch_history(ticker, period="2y")
    print(f"  got {len(df)} bars, range {df['timestamps'].iloc[0].date()} "
          f"to {df['timestamps'].iloc[-1].date()}")
    print(f"  last close: ${df['close'].iloc[-1]:.2f}")

    # 2. Load the model
    print("\nStep 2/4: loading Kronos-small (~95 MB first-run download)...")
    wrapper = load_kronos(model_id="small", device="cpu")
    info = wrapper.info()
    print(f"  model:    {info['model_id']}")
    print(f"  context:  {info['max_context']} bars")
    print(f"  device:   {info['device']}")

    # 3. Build the prediction-time index
    print("\nStep 3/4: building the 30-day forecast horizon...")
    last_ts = df["timestamps"].iloc[-1]
    y_ts = build_future_index(last_ts, periods=30, freq="1D")
    print(f"  forecasting {y_ts[0].date()} through {y_ts[-1].date()}")

    # 4. Run the prediction
    print("\nStep 4/4: running predict() with T=1.0, top_p=0.9, sample_count=1...")
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]],
        x_timestamp=df["timestamps"],
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=30,
        T=1.0,           # temperature: 1.0 is neutral
        top_p=0.9,       # nucleus sampling: 90% of probability mass
        sample_count=1,  # one path; increase for quantile bands
    )
    print(f"  forecast shape: {pred.shape}")
    print(f"  predicted next close: ${pred['close'].iloc[0]:.2f}")
    print(f"  predicted close in 30 days: ${pred['close'].iloc[-1]:.2f}")
    expected_ret = (pred["close"].iloc[-1] - df["close"].iloc[-1]) / df["close"].iloc[-1] * 100
    print(f"  expected 30-day return: {expected_ret:+.2f}%")

    print("\nForecast table (first 5 rows):")
    print(pred.head().to_string(index=False))
    print("\nForecast table (last 5 rows):")
    print(pred.tail().to_string(index=False))

    # 5. Plot
    out_path = Path("aapl_forecast.png")
    fig, ax = plt.subplots(figsize=(11, 5))
    df["close"].tail(120).plot(ax=ax, label="Historical close", color="#26a69a")
    pred["close"].plot(ax=ax, label="Forecast (30 days)", color="#7aa2f7", linestyle="--")
    ax.axvline(df["timestamps"].iloc[-1], color="#888", linestyle=":", linewidth=1)
    ax.text(df["timestamps"].iloc[-1], ax.get_ylim()[1] * 0.98, "  T+0",
            color="#888", fontsize=9, va="top")
    ax.set_title(f"{ticker}: 30-day forecast (Kronos-small, T=1.0, top_p=0.9)")
    ax.set_xlabel("Date")
    ax.set_ylabel("Close price ($)")
    ax.legend()
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig(out_path, dpi=120)
    print(f"\nSaved plot to {out_path}")

    # 6. Sanity checks (a few quick assertions so you know your env is sane)
    assert len(pred) == 30, f"expected 30 forecast rows, got {len(pred)}"
    assert pred["close"].isna().sum() == 0, "forecast contains NaNs"
    assert (pred["high"] >= pred["low"]).all(), "forecast has high < low"
    print("\nSanity checks: PASSED")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 02 — 02_us_stock_yfinance.py

Skill: Beginner. Theme: Multi-ticker US watchlist with overlay chart. Extra needed: [global].

Run: python 02_us_stock_yfinance.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 02 — US equities via yfinance, multi-ticker comparison.

WHAT THIS DEMONSTRATES
-----------------------
How to forecast multiple US tickers in one script and produce a side-by-side
comparison. This is the "weekly watchlist update" workflow — run it once a
week to see what Kronos thinks about your portfolio.

WHEN TO USE THIS
----------------
You follow a set of stocks (sector peers, watchlist, portfolio) and want a
single chart or table that compares Kronos's next-30-day prediction for each
one. Common for retail quants, sector analysts, financial journalists.

EXPECTED OUTPUT
---------------
- A printed table with one row per ticker: last close, predicted next close,
  30-day expected return, and a quick verdict ("bullish" / "bearish" /
  "flat") based on the predicted return.
- A PNG file `us_watchlist_forecast.png` with overlaid close-price traces.

COMMON PITFALLS
---------------
- **yfinance rate limit**: too many requests in a short window; the yfinance
  library will raise a YFRateLimitError. Sleep between calls or batch.
- **Ticker delisted**: returns empty DataFrame; we wrap each ticker in
  try/except and print a warning so one bad symbol doesn't kill the run.
- **Different lookback lengths**: yfinance returns different history depths
  depending on interval; we standardize with `tail(400)` so all tickers
  feed the same number of bars to the model.

BEFORE RUNNING
--------------
    pip install kronos-finance[global,analysis]
"""
from __future__ import annotations

import sys
import time
from pathlib import Path

import matplotlib.pyplot as plt
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def classify_expected_return(ret_pct: float, bull: float = 5.0, bear: float = -5.0) -> str:
    """Return a one-word verdict for a predicted 30-day return."""
    if ret_pct > bull:
        return "bullish"
    if ret_pct < bear:
        return "bearish"
    return "flat"


def forecast_one(wrapper, ticker: str, lookback: int = 400,
                 pred_len: int = 30, interval: str = "1d",
                 period: str = "1y") -> dict | None:
    """Forecast a single ticker. Returns a summary dict or None on failure."""
    try:
        df = load_ohlcv(ticker, period=period, interval=interval)
    except Exception as e:
        print(f"  WARN: {ticker}: data fetch failed: {e}")
        return None
    if len(df) < 60:
        print(f"  WARN: {ticker}: insufficient history ({len(df)} bars); skipping")
        return None

    x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(lookback)
    x_ts = df["timestamps"].tail(lookback)
    last_ts = df["timestamps"].iloc[-1]
    freq = "1D" if interval == "1d" else interval
    y_ts = pd.date_range(last_ts, periods=pred_len + 1, freq=freq)[1:]

    try:
        pred = wrapper.predict(
            df=x_df, x_timestamp=x_ts,
            y_timestamp=pd.Series(y_ts, name="timestamps"),
            pred_len=pred_len, T=1.0, top_p=0.9, sample_count=1,
        )
    except Exception as e:
        print(f"  WARN: {ticker}: predict failed: {e}")
        return None

    last_close = float(df["close"].iloc[-1])
    next_close = float(pred["close"].iloc[0])
    close_30d = float(pred["close"].iloc[-1])
    ret_next = (next_close - last_close) / last_close * 100
    ret_30d = (close_30d - last_close) / last_close * 100
    return {
        "ticker": ticker,
        "last_close": last_close,
        "pred_next": next_close,
        "pred_close_30d": close_30d,
        "ret_next_pct": ret_next,
        "ret_30d_pct": ret_30d,
        "verdict": classify_expected_return(ret_30d),
        "pred_df": pred,
        "hist_df": df.tail(60),  # only keep what we plot
    }


def main() -> int:
    watchlist = ["AAPL", "MSFT", "NVDA", "AMZN", "GOOGL", "META", "TSLA"]
    print(f"=== US watchlist forecast ({len(watchlist)} tickers) ===\n")

    # Load the model once — predictions are the slow part; reuse it
    print("Loading Kronos-small...")
    wrapper = load_kronos(model_id="small", device="cpu")

    results: list[dict] = []
    for ticker in watchlist:
        print(f"  fetching + forecasting {ticker}...")
        r = forecast_one(wrapper, ticker)
        if r is not None:
            results.append(r)
        # Tiny sleep to be polite to yfinance (helps avoid rate limits
        # on a fresh run; comment out if you're impatient)
        time.sleep(0.2)

    if not results:
        print("No results to show.")
        return 1

    # Summary table
    print("\n=== Forecast summary ===")
    summary_rows = [
        {
            "ticker": r["ticker"],
            "last_close": f"${r['last_close']:.2f}",
            "next_close": f"${r['pred_next']:.2f}",
            "next_ret": f"{r['ret_next_pct']:+.2f}%",
            "30d_close": f"${r['pred_close_30d']:.2f}",
            "30d_ret": f"{r['ret_30d_pct']:+.2f}%",
            "verdict": r["verdict"],
        }
        for r in results
    ]
    summary = pd.DataFrame(summary_rows)
    print(summary.to_string(index=False))

    # Save the table to CSV
    out_csv = Path("us_watchlist_forecast.csv")
    summary.to_csv(out_csv, index=False)
    print(f"\nSaved summary table to {out_csv}")

    # Overlay plot
    fig, ax = plt.subplots(figsize=(12, 6))
    colors = ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728", "#9467bd",
              "#8c564b", "#e377c2"]
    for i, r in enumerate(results):
        hist = r["hist_df"]
        pred = r["pred_df"]
        # Use a numeric index so all tickers align on the same x-axis
        hist_idx = range(len(hist))
        pred_idx = range(len(hist) - 1, len(hist) - 1 + len(pred))
        color = colors[i % len(colors)]
        ax.plot(list(hist_idx), hist["close"].values, color=color, label=r["ticker"])
        ax.plot(list(pred_idx), pred["close"].values, color=color, linestyle="--", alpha=0.7)

    ax.axvline(len(results[0]["hist_df"]) - 1, color="#888", linestyle=":")
    ax.set_title("US watchlist: 30-day forecast (dashed = forecast)")
    ax.set_xlabel("Bars from end of history")
    ax.set_ylabel("Close price ($)")
    ax.legend(loc="upper left", ncol=4)
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    out_png = Path("us_watchlist_forecast.png")
    plt.savefig(out_png, dpi=120)
    print(f"Saved overlay plot to {out_png}")

    # Quick stats
    bullish = sum(1 for r in results if r["verdict"] == "bullish")
    bearish = sum(1 for r in results if r["verdict"] == "bearish")
    flat = sum(1 for r in results if r["verdict"] == "flat")
    print(f"\nAggregate verdict: {bullish} bullish, {flat} flat, {bearish} bearish")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 03 — 03_cn_a_share_akshare.py

Skill: Beginner. Theme: CN A-share with column-rename walkthrough. Extra needed: [cn].

Run: python 03_cn_a_share_akshare.py (after pip install kronos-finance[cn] if extra != (none))

Full source:

"""Example 03 — CN A-share via AKShare, with column-rename and lunch-break handling.

WHAT THIS DEMONSTRATES
-----------------------
Three things that make CN market forecasting different from US:
  1. AKShare returns OHLCV columns in Chinese; kronos_finance.load_ohlcv()
     handles the rename automatically, but we also show the manual path.
  2. CN trading hours have a lunch break (11:30-13:00) — for daily bars
     this doesn't matter, but for intraday you need to be aware of gaps.
  3. CN A-share tickers are 6-digit numeric strings; we walk through the
     auto-detection rules so you know how AKShare got picked.

WHEN TO USE THIS
----------------
You're forecasting Chinese A-shares (沪深京), indexes (沪深300), or any
CN instrument that AKShare covers. Also a good reference if you're seeing
"empty dataframe" errors from AKShare — we cover the 4 most common causes.

EXPECTED OUTPUT
---------------
- 600+ bars of 600519 (Kweichow Moutai) daily history.
- A 30-day forecast table.
- A line chart showing historical close + forecast trace.

COMMON PITFALLS
---------------
- **AKShare column names**: AKShare returns 日期/开盘/最高/最低/收盘/成交量/
  成交额. kronos_finance normalizes them; if you call ak.stock_zh_a_hist
  directly you must rename.
- **Adjusted vs unadjusted prices**: AKShare defaults to adjusted when you
  pass `adjust="qfq"`; we pass that explicitly. Don't train on raw prices
  or splits will create fake crashes.
- **Empty results**: AKShare returns empty for ETFs (uses different API),
  indexes, or recently-listed tickers. Use the source picker or the
  fallback chain in the code.
- **Lunch-break gaps in minute bars**: AKShare's 5-minute bars have gaps
  from 11:30 to 13:00. For daily bars this is invisible. For minute
  bars, resample to 30-min or 60-min first.

BEFORE RUNNING
--------------
    pip install kronos-finance[cn]
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib.pyplot as plt
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import _standardize, load_ohlcv  # noqa: E402


def explain_source_detection(ticker: str) -> str:
    """Walk through why _guess_source() picks a particular loader.

    Educational — helps you debug "wrong source" issues.
    """
    if "/" in ticker:
        return "ccxt (slash in ticker = crypto pair)"
    if ticker.upper().startswith(("SH", "SZ")) and ticker[2:].isdigit():
        return "qlib (SH/SZ prefix + digits = CN quant platform)"
    if ticker.isdigit() and len(ticker) == 6:
        return "akshare (6 digits = CN A-share)"
    if ticker.upper().endswith((".SS", ".SZ", ".HK", ".T", ".L")):
        return "yfinance (Yahoo-style suffix = global equity)"
    return "yfinance (default fallback)"


def show_manual_rename() -> None:
    """Show what load_ohlcv does behind the scenes."""
    print("\n--- Behind the scenes ---")
    print("AKShare returns columns: 日期, 开盘, 最高, 最低, 收盘, 成交量, 成交额")
    print("kronos_finance.load_ohlcv() renames them to:")
    print("  timestamps, open, high, low, close, volume, amount")
    print("If you call AKShare directly, you'd write:")
    print("  df = ak.stock_zh_a_hist(symbol=ticker, period='daily', adjust='qfq')")
    print("  df = df.rename(columns={'日期': 'timestamps', '开盘': 'open', ...})")
    print("  df = _standardize(df)  # tz-naive, sorted, amount = close * volume")
    print("Then you can pass df to wrapper.predict() directly.\n")


def main() -> int:
    ticker = "600519"  # Kweichow Moutai, the canonical CN A-share example
    print(f"=== CN A-share forecast: {ticker} (Kweichow Moutai) ===\n")

    # Educational: show the source-detection logic
    print(f"Why does load_ohlcv('{ticker}') use AKShare?")
    print(f"  -> {explain_source_detection(ticker)}")
    show_manual_rename()

    # 1. Fetch via kronos_finance (the supported path)
    print(f"Step 1/3: fetching {ticker} daily bars via AKShare...")
    df = load_ohlcv(ticker, source="akshare", period="daily")
    print(f"  got {len(df)} bars, range {df['timestamps'].iloc[0].date()} "
          f"to {df['timestamps'].iloc[-1].date()}")
    print(f"  last close: ¥{df['close'].iloc[-1]:.2f}")
    print(f"  columns: {list(df.columns)}")
    assert list(df.columns) == ["timestamps", "open", "high", "low", "close", "volume", "amount"]

    # 2. Demonstrate manual fetch path
    print("\nStep 2/3: demonstration — fetch directly via AKShare and standardize...")
    try:
        import akshare as ak
        raw = ak.stock_zh_a_hist(symbol=ticker, period="daily", adjust="qfq")
        print(f"  raw AKShare columns: {list(raw.columns)}")
        renamed = raw.rename(columns={
            "日期": "timestamps", "开盘": "open", "最高": "high",
            "最低": "low", "收盘": "close",
            "成交量": "volume", "成交额": "amount",
        })
        standardized = _standardize(renamed)
        print(f"  after rename + standardize: {list(standardized.columns)}")
        assert list(standardized.columns) == list(df.columns), "manual path differs from helper!"
        print("  manual path produces identical DataFrame to load_ohlcv() ✓")
    except ImportError:
        print("  akshare not installed; skipping manual demonstration")
    except Exception as e:
        print(f"  manual path failed: {e}")

    # 3. Run the forecast
    print("\nStep 3/3: running predict()...")
    wrapper = load_kronos(model_id="small", device="cpu")
    y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
        x_timestamp=df["timestamps"].tail(400),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=30,
    )

    last_close = df["close"].iloc[-1]
    next_close = pred["close"].iloc[0]
    close_30d = pred["close"].iloc[-1]
    print(f"  predicted next close: ¥{next_close:.2f}")
    print(f"  predicted 30-day close: ¥{close_30d:.2f}")
    print(f"  expected 30-day return: {(close_30d - last_close) / last_close * 100:+.2f}%")

    # Save to CSV (use ¥ in filename for clarity)
    out_csv = Path(f"{ticker}_forecast.csv")
    pred.to_csv(out_csv, index=False)
    print(f"\nSaved forecast to {out_csv}")

    # Plot
    out_png = Path(f"{ticker}_forecast.png")
    fig, ax = plt.subplots(figsize=(11, 5))
    df["close"].tail(120).plot(ax=ax, label="Historical close (¥)", color="#d62728")
    pred["close"].plot(ax=ax, label="Forecast (30 days, ¥)", color="#1f77b4", linestyle="--")
    ax.axvline(df["timestamps"].iloc[-1], color="#888", linestyle=":")
    ax.set_title(f"{ticker} (Kweichow Moutai): 30-day forecast")
    ax.set_xlabel("Date")
    ax.set_ylabel("Close price (¥)")
    ax.legend()
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig(out_png, dpi=120)
    print(f"Saved plot to {out_png}")

    print("\nNote: for intraday CN data, AKShare minute bars have a lunch")
    print("break gap (11:30-13:00). Resample to 30-min or 60-min to flatten it.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 04 — 04_crypto_ccxt.py

Skill: Beginner. Theme: BTC/USDT 5-minute bars from Binance. Extra needed: [crypto].

Run: python 04_crypto_ccxt.py (after pip install kronos-finance[crypto] if extra != (none))

Full source:

"""Example 04 — Crypto via CCXT (Binance BTC/USDT 5-minute bars).

WHAT THIS DEMONSTRATES
-----------------------
Crypto-specific forecasting workflow:
  1. Use CCXT to pull 5-minute OHLCV bars from Binance (or any of 100+
     exchanges).
  2. Handle the volume-scale issue (BTC volume vs altcoin volume differ by
     1000x — keep them on different model runs).
  3. Short forecast horizon (120 bars = 10 hours) and what that means for
     trading decisions.
  4. The difference between using T=1.0 (sampling) vs T=0.0 (deterministic
     greedy) for high-frequency crypto prediction.

WHEN TO USE THIS
----------------
You trade crypto intraday and want a near-term price-direction signal.
Not for swing-trading (use daily bars + 7-day forecast instead). Not for
long-term investing (Kronos isn't trained for that horizon).

EXPECTED OUTPUT
---------------
- 500 5-minute bars of BTC/USDT (about 42 hours of history).
- A 120-bar (10-hour) forecast.
- Sanity checks: forecast open ~ historical close, no NaNs, etc.

COMMON PITFALLS
---------------
- **Rate limits**: Binance allows ~1200 requests/min for free tier; CCXT
  enables rate-limiting by default. If you hit limits, sleep between calls.
- **Symbol format**: Binance uses "BTC/USDT" (slash). Other exchanges use
  different formats: Kraken "XBT/USD", Coinbase "BTC-USD". CCXT abstracts
  this somewhat.
- **24/7 markets**: crypto trades every day. yfinance-style "trading hours"
  don't apply. The model's day-of-week features may not be meaningful for
  crypto at 5-min granularity.
- **Volume scale**: BTC trades ~$20B/day, small altcoins trade $10K/day.
  Don't mix them in the same training run.
- **Funding rates / liquidations**: Kronos only sees OHLCV. Funding rate
  spikes and liquidation cascades aren't in the input and will look like
  random shocks to the model.

BEFORE RUNNING
--------------
    pip install kronos-finance[crypto]
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib.pyplot as plt
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def main() -> int:
    symbol = "BTC/USDT"
    exchange = "binance"
    interval = "5m"
    lookback = 400      # 400 5-min bars = ~33 hours of context
    pred_len = 120      # 120 5-min bars = 10 hours of forecast
    print(f"=== Crypto forecast: {symbol} on {exchange} ({interval} bars) ===\n")

    # 1. Fetch
    print(f"Step 1/3: fetching {lookback} {interval} bars from {exchange}...")
    df = load_ohlcv(symbol, source="ccxt", exchange=exchange,
                    interval=interval, limit=lookback)
    print(f"  got {len(df)} bars, range {df['timestamps'].iloc[0]} "
          f"to {df['timestamps'].iloc[-1]}")
    print(f"  last close: ${df['close'].iloc[-1]:,.2f}")
    print(f"  last volume: {df['volume'].iloc[-1]:,.2f} BTC")

    # Sanity: check for gaps in the 5-min series (weekends are fine, but
    # look for unexpected gaps from API hiccups)
    expected_freq = pd.Timedelta("5min")
    gaps = df["timestamps"].diff().dropna()
    big_gaps = gaps[gaps > 2 * expected_freq]
    if len(big_gaps) > 0:
        print(f"  warning: {len(big_gaps)} gaps > 10 minutes found:")
        for ts, gap in big_gaps.head(3).items():
            print(f"    at index {ts}: gap of {gap}")

    # 2. Build forecast index
    print(f"\nStep 2/3: building {pred_len}-bar ({pred_len * 5 / 60:.1f}h) horizon...")
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=pred_len + 1, freq="5min")[1:]

    # 3. Predict
    # Use T=0.5 for crypto — slightly more conservative than neutral.
    # For high-frequency, lower temperature gives less jitter in the
    # predicted sequence (which often translates to less false signals).
    print(f"\nStep 3/3: running predict() with T=0.5, top_p=0.9...")
    wrapper = load_kronos(model_id="small", device="cpu")
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(lookback),
        x_timestamp=df["timestamps"].tail(lookback),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=pred_len,
        T=0.5,           # slightly more conservative
        top_p=0.9,
        sample_count=1,
    )

    last_close = df["close"].iloc[-1]
    next_close = pred["close"].iloc[0]
    close_10h = pred["close"].iloc[-1]
    print(f"  predicted next-bar close: ${next_close:,.2f}")
    print(f"  predicted 10-hour close:  ${close_10h:,.2f}")
    print(f"  expected 10-hour return:  {(close_10h - last_close) / last_close * 100:+.3f}%")

    # Sanity checks
    assert len(pred) == pred_len, f"expected {pred_len} rows, got {len(pred)}"
    assert pred["close"].isna().sum() == 0, "forecast contains NaNs"
    assert (pred["high"] >= pred["low"]).all(), "high < low in forecast"
    assert (pred["high"] >= pred["close"]).all(), "high < close in forecast"
    assert (pred["low"] <= pred["close"]).all(), "low > close in forecast"
    print("  sanity checks: PASSED")

    # Save
    out_csv = Path("btcusdt_10h_forecast.csv")
    pred.to_csv(out_csv, index=False)
    print(f"\nSaved forecast to {out_csv}")

    # Plot
    out_png = Path("btcusdt_10h_forecast.png")
    fig, ax = plt.subplots(figsize=(12, 5))
    df["close"].tail(120).plot(ax=ax, label="Historical 5m close", color="#f7931a")
    pred["close"].plot(ax=ax, label=f"Forecast ({pred_len * 5 / 60:.0f}h)",
                       color="#1f77b4", linestyle="--")
    ax.axvline(df["timestamps"].iloc[-1], color="#888", linestyle=":")
    ax.set_title(f"{symbol} — {pred_len}-bar forecast on {exchange}")
    ax.set_xlabel("Time (UTC)")
    ax.set_ylabel("Close price ($)")
    ax.legend()
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig(out_png, dpi=120)
    print(f"Saved plot to {out_png}")

    # Footnote on high-frequency caveats
    print("\nNOTE: 5-minute crypto forecasts are noise-sensitive. For real")
    print("trading decisions, ensemble multiple sample paths (set")
    print("sample_count=5 in wrapper.predict()) and compare to BTC dominance,")
    print("funding rates, and orderbook imbalance — none of which Kronos sees.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 05 — 05_batch_predict.py

Skill: Intermediate. Theme: Batch sweep across many tickers. Extra needed: [global].

Run: python 05_batch_predict.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 05 — Batch prediction across many tickers with progress reporting.

WHAT THIS DEMONSTRATES
-----------------------
How to scale kronos_finance to hundreds or thousands of tickers. Shows:
  1. A robust batch loop that survives per-ticker failures.
  2. Progress reporting via tqdm (or a simple fallback if tqdm not available).
  3. Concurrency vs sequential trade-offs (we stay sequential for predictability).
  4. Writing a single big CSV in a memory-efficient way.
  5. Aggregating per-ticker forecasts into a summary DataFrame.

WHEN TO USE THIS
----------------
You maintain a watchlist (50+ tickers), run sweeps over universes (S&P 500,
CSI 300, top 100 coins by volume), or you want to build a market-summary
report once a day.

EXPECTED OUTPUT
---------------
- `batch_predictions_long.csv` — long-format CSV (one row per
  ticker × forecast bar) for loading into a database or pandas.
- `batch_summary.csv` — wide-format summary with one row per ticker.
- Console output: progress + per-ticker status.

PERFORMANCE
-----------
On a CPU, ~3-10 seconds per ticker for a 30-day forecast. On a CUDA GPU
(RTX 4070 or better), ~0.05-0.2 seconds per ticker. For a 500-ticker
sweep: ~25-50 minutes CPU, ~5-10 minutes GPU.

COMMON PITFALLS
---------------
- **Memory**: long-format CSV is fine; wide-format with 30 cols per ticker
  × 500 tickers × 3650 bars per ticker is huge. Use long format for batch
  and let pandas reshape as needed.
- **Rate limiting**: data sources (yfinance, AKShare, CCXT) all have rate
  limits. The helper sleeps 0.2s between calls. Lower for paid data feeds,
  raise for free accounts.
- **Mixed currency outputs**: don't compare absolute prices across tickers
  in different currencies. Use percent returns for cross-section analysis.
- **Yanked tickers**: always try/except around per-ticker work; one bad
  ticker should never kill the sweep.

BEFORE RUNNING
--------------
    pip install kronos-finance[global,analysis]
"""
from __future__ import annotations

import sys
import time
from datetime import date
from pathlib import Path

import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


# Tiny progress-bar fallback so we don't take a tqdm dependency
class ProgressCounter:
    def __init__(self, total: int, label: str = "progress") -> None:
        self.total = total
        self.n = 0
        self.label = label

    def update(self) -> None:
        self.n += 1
        pct = self.n / self.total * 100
        bar = "█" * int(pct // 2) + " " * int(50 - pct // 2)
        print(f"\r  {self.label} [{bar}] {self.n}/{self.total} ({pct:.0f}%)",
              end="", flush=True)

    def done(self) -> None:
        print()


def forecast_one(wrapper, ticker: str, pred_len: int = 30) -> dict | None:
    """Run a forecast for one ticker and return row data. Returns None on any failure."""
    try:
        df = load_ohlcv(ticker, period="1y", interval="1d")
    except Exception as e:
        print(f"  [WARN] {ticker}: data fetch failed ({e.__class__.__name__})")
        return None
    if len(df) < 60:
        print(f"  [WARN] {ticker}: insufficient history ({len(df)} bars)")
        return None

    x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(400)
    x_ts = df["timestamps"].tail(400)
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=pred_len + 1, freq="1D")[1:]

    try:
        pred = wrapper.predict(
            df=x_df, x_timestamp=x_ts,
            y_timestamp=pd.Series(y_ts, name="timestamps"),
            pred_len=pred_len, T=1.0, top_p=0.9, sample_count=1,
        )
    except Exception as e:
        print(f"  [WARN] {ticker}: predict failed ({e.__class__.__name__}: {e})")
        return None

    last_close = float(df["close"].iloc[-1])
    next_close = float(pred["close"].iloc[0])
    close_30d = float(pred["close"].iloc[-1])
    ret_30d = (close_30d - last_close) / last_close * 100
    return {
        "ticker": ticker,
        "last_close": last_close,
        "pred_next": next_close,
        "pred_close_30d": close_30d,
        "ret_30d_pct": ret_30d,
    }


def main() -> int:
    # A small universe — 10 well-known tickers from different markets.
    # For a 100+ ticker sweep, just expand this list.
    universe = [
        "AAPL", "MSFT", "NVDA", "AMZN", "GOOGL",  # US tech
        "600519", "000001", "600036",              # CN banks & consumer
        "BTC/USDT",                                # crypto
    ]

    print(f"=== Batch prediction sweep: {len(universe)} tickers ===")
    print(f"Date: {date.today().isoformat()}\n")

    print("Loading Kronos-small...")
    wrapper = load_kronos(model_id="small", device="cpu")
    print()

    progress = ProgressCounter(total=len(universe), label="tick")
    summary_rows: list[dict] = []
    pred_records: list[dict] = []

    for ticker in universe:
        result = forecast_one(wrapper, ticker)
        progress.update()
        if result is None:
            continue
        summary_rows.append(result)

        # Re-run predict to get the long-format records for export
        # (We could store the pred DataFrame inside forecast_one to avoid
        # the re-run, but keeping it small here for readability.)
        try:
            df = load_ohlcv(ticker, period="1y", interval="1d")
            x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(400)
            x_ts = df["timestamps"].tail(400)
            last_ts = df["timestamps"].iloc[-1]
            y_ts = pd.date_range(last_ts, periods=31, freq="1D")[1:]
            pred = wrapper.predict(
                df=x_df, x_timestamp=x_ts,
                y_timestamp=pd.Series(y_ts, name="timestamps"),
                pred_len=30, T=1.0, top_p=0.9, sample_count=1,
            )
            for ts, row in zip(y_ts, pred.to_dict("records")):
                pred_records.append({
                    "ticker": ticker,
                    "timestamps": ts,
                    **{k: v for k, v in row.items()},
                })
        except Exception as e:
            print(f"  [WARN] {ticker}: long-format re-run failed ({e})")
        time.sleep(0.1)

    progress.done()

    if not summary_rows:
        print("No successful predictions.")
        return 1

    # Summary
    summary = pd.DataFrame(summary_rows)
    summary = summary.sort_values("ret_30d_pct", ascending=False)
    summary["rank"] = range(1, len(summary) + 1)
    summary = summary[["rank", "ticker", "last_close",
                       "pred_next", "pred_close_30d", "ret_30d_pct"]]
    print("\n=== Summary (sorted by 30-day expected return) ===\n")
    print(summary.to_string(index=False, float_format=lambda x: f"{x:.3f}"))

    out_summary = Path("batch_summary.csv")
    summary.to_csv(out_summary, index=False)
    print(f"\nSaved summary -> {out_summary}")

    # Long-format detailed predictions
    if pred_records:
        long_df = pd.DataFrame(pred_records)
        # Reorder columns for friendliness
        cols = ["ticker", "timestamps", "open", "high", "low", "close", "volume", "amount"]
        long_df = long_df[cols]
        out_long = Path("batch_predictions_long.csv")
        long_df.to_csv(out_long, index=False)
        print(f"Saved long-format predictions -> {out_long} ({len(long_df)} rows)")

    # Aggregate stats
    returns = summary["ret_30d_pct"]
    print(f"\nAggregate stats over {len(summary)} tickers:")
    print(f"  mean expected return: {returns.mean():+.2f}%")
    print(f"  median expected return: {returns.median():+.2f}%")
    print(f"  std (cross-section): {returns.std():.2f}%")
    print(f"  range: [{returns.min():+.2f}%, {returns.max():+.2f}%]")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 06 — 06_indicators_and_tearsheet.py

Skill: Intermediate. Theme: RSI/MACD/Bollinger + quantstats tearsheet. Extra needed: [global,analysis].

Run: python 06_indicators_and_tearsheet.py (after pip install kronos-finance[global,analysis] if extra != (none))

Full source:

"""Example 06 — Indicators, signals, and a Tearsheet (quantstats HTML).

WHAT THIS DEMONSTRATES
-----------------------
Three pillars of turning a forecast into something actionable:
  1. Compute technical indicators (RSI, MACD, Bollinger bands) on the
     historical DataFrame using kronos_finance.analysis.enrich_with_indicators.
  2. Convert the forecast into a trading return series via
     forecast_to_returns(threshold=...).
  3. Build a quantstats HTML tearsheet comparing the strategy vs buy-and-hold.

WHEN TO USE THIS
----------------
You have a forecast and want to make a trading decision out of it, or you
want to evaluate the quality of past Kronos predictions against realized
outcomes. Quantstats tearsheets are the industry-standard format for that
evaluation.

EXPECTED OUTPUT
---------------
- A CSV with the historical bars + indicator columns (`ticker_indicators.csv`).
- A CSV with the strategy's daily returns (`ticker_strategy_returns.csv`).
- An HTML tearsheet (`tearsheet.html`) — open in any browser; it has 30+
  panels (Sharpe, Sortino, drawdown, monthly heatmap, etc.).

COMMON PITFALLS
---------------
- **Tearsheet on forecast-only returns is meaningless** for evaluation.
  For a real evaluation, you need historical Kronos predictions vs
  realized outcomes. The tearsheet here illustrates the API, not a real
  backtest.
- **Missing analysis extras**: `enrich_with_indicators` needs pandas-ta,
  `make_tearsheet` needs quantstats. Install with `[analysis]` extra.
- **Threshold sensitivity**: forecast_to_returns(threshold=0.02) will
  produce zero positions if the forecast never beats +2%. Tune the
  threshold to your risk appetite.

BEFORE RUNNING
--------------
    pip install kronos-finance[global,analysis]
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib.dates as mdates
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.analysis import (  # noqa: E402
    enrich_with_indicators, forecast_to_returns, make_tearsheet,
)
from kronos_finance.data import load_ohlcv  # noqa: E402


def plot_price_and_forecast(history_close: pd.Series,
                            forecast_close: pd.Series,
                            out_path: Path) -> None:
    """Quick chart: history vs forecast."""
    fig, ax = plt.subplots(figsize=(11, 5))
    history_close.plot(ax=ax, label="History", color="#26a69a", linewidth=1)
    forecast_close.plot(ax=ax, label="Forecast (30 days)", color="#7aa2f7",
                        linewidth=1, linestyle="--")
    ax.axvline(history_close.index[-1], color="#888", linestyle=":")
    ax.set_title("Historical close + forecast")
    ax.set_xlabel("Date")
    ax.set_ylabel("Close ($)")
    ax.legend()
    ax.grid(True, alpha=0.3)
    ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))
    plt.setp(ax.get_xticklabels(), rotation=30, ha="right")
    plt.tight_layout()
    plt.savefig(out_path, dpi=120)
    print(f"Saved price+forecast chart -> {out_path}")


def plot_indicators(combined: pd.DataFrame, out_path: Path) -> None:
    """Three-panel: price+Bollinger, RSI, MACD."""
    fig, axes = plt.subplots(3, 1, figsize=(12, 10),
                             gridspec_kw={"height_ratios": [3, 1, 1]},
                             sharex=False)

    # Panel 1: Close with Bollinger bands
    ax = axes[0]
    ax.plot(combined.index, combined["close"], color="#26a69a", label="Close")
    if "BBU_20_2.0" in combined.columns and "BBL_20_2.0" in combined.columns:
        ax.plot(combined.index, combined["BBU_20_2.0"], color="#1f77b4",
                linestyle="--", label="BB upper", alpha=0.7)
        ax.plot(combined.index, combined["BBM_20_2.0"], color="#888",
                linestyle="--", label="BB middle", alpha=0.7)
        ax.plot(combined.index, combined["BBL_20_2.0"], color="#d62728",
                linestyle="--", label="BB lower", alpha=0.7)
    ax.set_title("Bollinger bands")
    ax.set_ylabel("Close ($)")
    ax.legend(loc="upper left", fontsize=8)
    ax.grid(True, alpha=0.3)

    # Panel 2: RSI
    ax = axes[1]
    if "RSI_14" in combined.columns:
        ax.plot(combined.index, combined["RSI_14"], color="#9c27b0", label="RSI(14)")
    ax.axhline(70, color="#d62728", linestyle=":", linewidth=1)
    ax.axhline(30, color="#26a69a", linestyle=":", linewidth=1)
    ax.fill_between(combined.index, 30, 70, alpha=0.05, color="#888")
    ax.set_ylim(0, 100)
    ax.set_ylabel("RSI")
    ax.grid(True, alpha=0.3)

    # Panel 3: MACD
    ax = axes[2]
    if "MACD_12_26_9" in combined.columns:
        ax.plot(combined.index, combined["MACD_12_26_9"], color="#1f77b4", label="MACD")
        ax.plot(combined.index, combined["MACDs_12_26_9"], color="#ff7f0e",
                label="Signal", linewidth=1)
        ax.bar(combined.index, combined["MACDh_12_26_9"],
               color=["#26a69a" if v >= 0 else "#d62728"
                      for v in combined["MACDh_12_26_9"]],
               alpha=0.5, width=0.8, label="Histogram")
    ax.axhline(0, color="#888", linewidth=0.5)
    ax.set_ylabel("MACD")
    ax.legend(loc="upper left", fontsize=8)
    ax.grid(True, alpha=0.3)

    for ax in axes:
        ax.xaxis.set_major_formatter(mdates.DateFormatter("%Y-%m-%d"))
    plt.setp(axes[-1].get_xticklabels(), rotation=30, ha="right")
    plt.tight_layout()
    plt.savefig(out_path, dpi=120)
    print(f"Saved indicator chart -> {out_path}")


def main() -> int:
    ticker = "AAPL"
    print(f"=== Indicators + Tearsheet: {ticker} ===\n")

    # 1. Fetch
    print(f"Fetching {ticker}...")
    df = load_ohlcv(ticker, period="2y", interval="1d")
    print(f"  fetched {len(df)} bars")

    # 2. Indicators
    print("Adding indicators (RSI, MACD, Bollinger) on historical bars...")
    enriched = enrich_with_indicators(df)
    print(f"  indicator columns added: "
          f"{[c for c in enriched.columns if c not in df.columns][:6]}...")
    out_csv = Path(f"{ticker}_indicators.csv")
    enriched.to_csv(out_csv, index=False)
    print(f"  saved -> {out_csv}")

    # 3. Forecast
    print("\nLoading model and forecasting 30 days...")
    wrapper = load_kronos(model_id="small", device="cpu")
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=31, freq="1D")[1:]
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
        x_timestamp=df["timestamps"].tail(400),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=30,
    )

    # 4. Visual: price + forecast
    plot_price_and_forecast(
        history_close=pd.Series(df["close"].tail(120).values,
                                index=pd.to_datetime(df["timestamps"].tail(120).values)),
        forecast_close=pd.Series(pred["close"].values,
                                 index=y_ts),
        out_path=Path(f"{ticker}_price_forecast.png"),
    )

    # 5. Visual: indicators (uses the historical enriched frame)
    # Reindex on a DatetimeIndex so matplotlib can format the x-axis
    enriched_idx = enriched.set_index(pd.to_datetime(enriched["timestamps"]))
    plot_indicators(enriched_idx.tail(180), out_path=Path(f"{ticker}_indicators.png"))

    # 6. Convert forecast into a return series via the helper
    print("\nConverting forecast to a return series...")
    last_close = float(df["close"].iloc[-1])
    strat_returns = forecast_to_returns(pred, last_close=last_close, threshold=0.0)
    ret_csv = Path(f"{ticker}_strategy_returns.csv")
    pd.DataFrame({"return": strat_returns}).to_csv(ret_csv)
    print(f"  strategy return mean: {strat_returns.mean():.5f}")
    print(f"  strategy return std:  {strat_returns.std():.5f}")
    print(f"  Sharpe (rough):       {(strat_returns.mean() / strat_returns.std() * np.sqrt(252)):.2f}")
    print(f"  saved -> {ret_csv}")

    # 7. Quantstats HTML tearsheet (vs buy-and-hold on the forecast period)
    print("\nGenerating quantstats HTML tearsheet...")
    # Build a buy-and-hold benchmark series for the same 30-day window
    fc_returns = pred["close"].pct_change().fillna(0).reset_index(drop=True)
    strat_returns = strat_returns.reset_index(drop=True)
    try:
        make_tearsheet(
            returns=strat_returns,
            benchmark=fc_returns,
            output=Path("tearsheet.html"),
        )
        print("  saved -> tearsheet.html (open in browser)")
    except Exception as e:
        print(f"  quantstats failed: {e}")
        print("  install with: pip install kronos-finance[analysis]")

    print("\nNOTE: tearsheets on forecast-only data are illustrative.")
    print("For real evaluation, fold historical Kronos predictions with")
    print("realized outcomes (walkforward testing) — not realized vs in-sample.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 07 — 07_cli_predict.sh

Skill: Beginner. Theme: kronos predict from a shell script. Extra needed: [global].

Run: bash 07_cli_predict.sh (after pip install kronos-finance[global] if extra != (none))

Full source:

#!/usr/bin/env bash
# Example 07 — CLI: one-shot prediction for a single ticker.
#
# WHAT THIS DEMONSTRATES
# -----------------------
# The `kronos` CLI can do predict without writing any Python. This is the
# "shell script you can put in cron" workflow. Run from any environment
# where kronos-finance is installed (pip install kronos-finance[global]).
#
# WHEN TO USE THIS
# ----------------
# You've installed kronos-finance and want to fire-and-forget a prediction
# from your shell or a cron job. Also useful for SSH-quick predictions
# on a remote box without setting up an editor.
#
# EXPECTED OUTPUT
# ---------------
# - A CSV at ./NVDA_forecast.csv with 30 rows of forecasted OHLCV.
# - A console table showing the first 5 forecast rows.
# - Exit code 0 on success, non-zero on failure (good for cron alerting).
#
# COMMON PITFALLS
# ---------------
# - The CLI is `kronos`, not `kronos-finance` (avoids name collision with
#   APT and other system packages).
# - First run downloads ~95 MB; subsequent runs are instant if the model
#   cache is present (~/.cache/huggingface/hub/...).
# - If your cron environment lacks HOME or PATH, set them explicitly:
#     export HOME=/home/user
#     export PATH=/home/user/.local/bin:$PATH
#
# USAGE
# -----
#   bash examples/07_cli_predict.sh
#
# Edit TICKER, PRED_LEN, EXPORT below to taste.

set -euo pipefail

TICKER="${TICKER:-NVDA}"
PRED_LEN="${PRED_LEN:-30}"
EXPORT="${EXPORT:-./NVDA_forecast.csv}"
MODEL="${MODEL:-small}"
DEVICE="${DEVICE:-cpu}"
INTERVAL="${INTERVAL:-1d}"

echo "=== kronos CLI: predict ${TICKER} -> ${EXPORT} ==="
echo "  model:    ${MODEL}"
echo "  device:   ${DEVICE}"
echo "  interval: ${INTERVAL}"
echo "  pred_len: ${PRED_LEN}"
echo ""

kronos predict "${TICKER}" \
    --model "${MODEL}" \
    --device "${DEVICE}" \
    --interval "${INTERVAL}" \
    --pred-len "${PRED_LEN}" \
    --export "${EXPORT}"

echo ""
echo "OK. Forecast saved to ${EXPORT}. First 5 rows:"
echo ""
python - <<PY
import pandas as pd
df = pd.read_csv("${EXPORT}")
print(df.head().to_string(index=False))
PY

---

Example 08 — 08_cli_backtest.sh

Skill: Intermediate. Theme: kronos backtest from a shell script. Extra needed: [global].

Run: bash 08_cli_backtest.sh (after pip install kronos-finance[global] if extra != (none))

Full source:

#!/usr/bin/env bash
# Example 08 — CLI: quick backtest for a single ticker.
#
# WHAT THIS DEMONSTRATES
# -----------------------
# The `kronos backtest` subcommand runs a simple historical Kronos
# prediction vs a buy-and-hold benchmark and exports the result. Useful
# for sanity-checking whether the model produces trading advantages on a
# given ticker over a given window.
#
# WHEN TO USE THIS
# ----------------
# You want to evaluate Kronos's effectiveness on a specific ticker over
# a specific historical window, without writing Python. Good for
# batch-comparing tickers in a shell loop.
#
# EXPECTED OUTPUT
# ---------------
# - A tearsheet (HTML or CSV, depending on the export flag) at the path
#   you pass to --export.
# - Console output of the headline metrics (Sharpe, total return,
#   drawdown, win rate).
#
# IMPORTANT CAVEATS
# -----------------
# The `kronos backtest` command is a *back-of-the-envelope* walk-forward
# sanity check, not a full event-driven backtest. For real evaluation,
# use the workflow shown in docs/BACKTESTING.md:
#   1. Predict from t-N to t-1.
#   2. Realize at t.
#   3. Compare.
#
# USAGE
# -----
#   bash examples/08_cli_backtest.sh

set -euo pipefail

TICKER="${TICKER:-AAPL}"
PERIOD="${PERIOD:-1y}"
INTERVAL="${INTERVAL:-1d}"
EXPORT="${EXPORT:-./AAPL_backtest_report.html}"
MODEL="${MODEL:-small}"
SOURCE="${SOURCE:-yfinance}"

echo "=== kronos CLI: backtest ${TICKER} over ${PERIOD} ==="
echo "  model:    ${MODEL}"
echo "  source:   ${SOURCE}"
echo "  interval: ${INTERVAL}"
echo "  period:   ${PERIOD}"
echo "  export:   ${EXPORT}"
echo ""

kronos backtest "${TICKER}" \
    --source "${SOURCE}" \
    --period "${PERIOD}" \
    --interval "${INTERVAL}" \
    --model "${MODEL}" \
    --export "${EXPORT}"

echo ""
echo "OK. Backtest report saved to ${EXPORT}."
echo "Open it in any browser for the full tearsheet."

# Tip: try a sweep over your watchlist
# for t in AAPL MSFT NVDA AMZN GOOGL; do
#     kronos backtest "$t" --export "./reports/${t}.html"
# done

---

Example 09 — 09_launch_dashboard.sh

Skill: Beginner. Theme: Start the Flask dashboard. Extra needed: [ui].

Run: bash 09_launch_dashboard.sh (after pip install kronos-finance[ui] if extra != (none))

Full source:

#!/usr/bin/env bash
# Example 09 — Launch the web dashboard.
#
# WHAT THIS DEMONSTRATES
# -----------------------
# `kronos ui` starts a Flask-based web UI in your terminal. Open the URL
# it prints and you get a chart-driven forecast interface — pick a
# ticker, choose a lookback and prediction horizon, click predict, see
# the chart update. No code.
#
# WHEN TO USE THIS
# ----------------
# You want to play with Kronos visually, share a dashboard with a
# non-coder colleague, or demo the model to a stakeholder. The UI is
# the friendliest entry point for anyone who doesn't want to write
# Python.
#
# EXPECTED OUTPUT
# ---------------
# - Console: "Running on http://127.0.0.1:5000" (or whatever you pass).
# - Browser tab with the dashboard open.
# - Ctrl-C to stop the server.
#
# DEPLOYMENT OPTIONS
# ------------------
# 1. Local (default): http://127.0.0.1:5000
# 2. LAN: `kronos ui --host 0.0.0.0` so other machines on your network
#    can hit it at http://<your-ip>:5000.
# 3. Production: behind a real WSGI server (gunicorn), see docs/DASHBOARD.md.
#
# COMMON PITFALLS
# ---------------
# - Port 5000 conflicts: macOS AirPlay uses 5000. Either kill it
#   (System Preferences -> Sharing -> AirPlay Receiver off) or pass
#   `--port 5001`.
# - "Permission denied" on a low port: keep --port >= 1024.
# - Browsers block 127.0.0.1 sometimes on first run; use http://localhost:5000.
#
# USAGE
# -----
#   bash examples/09_launch_dashboard.sh
#
# Override via env vars:
#   HOST=0.0.0.0 PORT=8080 bash examples/09_launch_dashboard.sh

set -euo pipefail

HOST="${HOST:-127.0.0.1}"
PORT="${PORT:-5000}"
DEBUG="${DEBUG:-false}"

echo "=== Launching kronos dashboard at http://${HOST}:${PORT} ==="
echo "  debug:  ${DEBUG}"
echo "  Open the URL in your browser. Ctrl-C to stop."
echo ""

kronos ui --host "${HOST}" --port "${PORT}" $([ "${DEBUG}" = "true" ] && echo "--debug")

---

Example 10 — 10_save_load_predictions.py

Skill: Intermediate. Theme: CSV / JSON / Parquet save/load round-trip. Extra needed: [global,analysis].

Run: python 10_save_load_predictions.py (after pip install kronos-finance[global,analysis] if extra != (none))

Full source:

"""Example 10 — Save, load, and concatenate predictions.

WHAT THIS DEMONSTRATES
-----------------------
Three patterns for working with predictions out-of-memory:
  1. Save a forecast to CSV.
  2. Save a forecast to JSON (for cross-language interop, e.g. JS dashboard).
  3. Load a forecast back and resume operations (concat with newer
     forecasts, compare with realized outcomes).
  4. Append predictions to a long-running history file (like a tick-store).

WHEN TO USE THIS
----------------
You're building a daily forecast pipeline. After each run you want to
append the new predictions to a master file (SQLite/Parquet) and then
later pull them back to compare against realized prices.

EXPECTED OUTPUT
---------------
- NVDA_forecast_20250923.csv (timestamped file)
- NVDA_forecast_20250923.json (same data, JSON for web)
- NVDA_history.parquet (master history file with this run appended)
- A printed comparison: forecast vs naive-baseline drift.

COMMON PITFALLS
---------------
- **CSV dtype loss**: volume as int + NaN can flip to float. Force the
  dtype on read or use parquet instead.
- **JSON date format**: pandas to_json uses epoch ms by default. We
  show the ISO-friendly format here.
- **Parquet dependency**: requires pyarrow. Install via the [analysis]
  extra (`pip install kronos-finance[analysis]`) or `pip install pyarrow`.

BEFORE RUNNING
--------------
    pip install kronos-finance[global,analysis]
"""
from __future__ import annotations

import json
import sys
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def main() -> int:
    ticker = "NVDA"
    print(f"=== Save / load / concat predictions: {ticker} ===\n")

    # 1. Fetch + forecast
    print("Fetching + forecasting...")
    df = load_ohlcv(ticker, period="1y", interval="1d")
    wrapper = load_kronos(model_id="small", device="cpu")
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=31, freq="1D")[1:]
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
        x_timestamp=df["timestamps"].tail(400),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=30,
    )

    # Tag with run metadata — invaluable when comparing runs
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    print(f"Run ID: {run_id}\n")
    pred_meta = pred.copy()
    pred_meta["ticker"] = ticker
    pred_meta["run_id"] = run_id
    pred_meta["model_id"] = "small"
    pred_meta["model_device"] = "cpu"

    # 2. Save to CSV with timestamped filename
    csv_path = Path(f"{ticker}_forecast_{run_id[:8]}.csv")
    pred_meta.to_csv(csv_path, index=False)
    print(f"CSV:    {csv_path} ({csv_path.stat().st_size} bytes)")
    print(f"  head: {list(pred_meta.columns)[:6]}...")

    # 3. Save to JSON (web-friendly, ISO timestamps)
    json_path = Path(f"{ticker}_forecast_{run_id[:8]}.json")
    json_obj = {
        "ticker": ticker,
        "run_id": run_id,
        "model_id": "small",
        "created_at": run_id,
        "history_last_close": float(df["close"].iloc[-1]),
        "forecast": json.loads(pred_meta.to_json(orient="records", date_format="iso")),
    }
    json_path.write_text(json.dumps(json_obj, indent=2))
    print(f"JSON:   {json_path} ({json_path.stat().st_size} bytes)")

    # 4. Parquet master file (append, dedupe on run_id+ticker+timestamp)
    pq_path = Path(f"{ticker}_history.parquet")
    if pq_path.exists():
        existing = pd.read_parquet(pq_path)
        combined = pd.concat([existing, pred_meta], ignore_index=True)
        # Dedupe: keep the newest run for a given (ticker, timestamp)
        combined = (combined
                    .sort_values("run_id")
                    .drop_duplicates(subset=["ticker", "timestamps"], keep="last"))
    else:
        combined = pred_meta
    combined.to_parquet(pq_path, index=False)
    print(f"PARQUET: {pq_path} (now {len(combined)} rows)")

    # 5. Load back the parquet, query
    print("\nReading back parquet and querying last 3 forecasts for this ticker...")
    reloaded = pd.read_parquet(pq_path)
    reloaded["timestamps"] = pd.to_datetime(reloaded["timestamps"])
    head_runs = (reloaded
                 .sort_values(["timestamps", "run_id"], ascending=[True, False])
                 .groupby("timestamps")
                 .head(1)
                 .tail(3))
    print(head_runs[["timestamps", "close", "run_id"]].to_string(index=False))

    # 6. Sanity: predict against the parquet master to load and re-predict
    # (Demonstrates round-tripping without re-fetching from yfinance)
    print("\nRound-trip test: load from parquet and re-forecast (skipped if")
    print("  data is too old). Use this pattern to validate your pipeline")
    print("  without hammering yfinance.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 11 — 11_qlib_finetune_smoke.py

Skill: Advanced. Theme: Qlib-format CSV for the upstream finetune script. Extra needed: [qlib].

Run: python 11_qlib_finetune_smoke.py (after pip install kronos-finance[qlib] if extra != (none))

Full source:

"""Example 11 — Qlib finetune smoke test.

WHAT THIS DEMONSTRATES
-----------------------
A walkthrough of how to:
  1. Convert a CSV of OHLCV into the Qlib binary format (qlib.init
     + dump_bin) so the upstream Kronos finetune script can ingest it.
  2. Build a minimal Python data object that mirrors what
     `examples/Kronos/finetune/qlib_exp/run.py` expects.
  3. Run a *very* small finetune (2 epochs on 1 ticker, tiny batch) to
     verify the pipeline works on your machine before scaling up.

WHEN TO USE THIS
----------------
You have ~6 months to multiple years of CN A-share daily data and want
to finetune Kronos-small on it. This example walks you through the
exact data preparation steps; the actual training script is in the
upstream Kronos repo (examples/Kronos/finetune/qlib_exp/run.py).

EXPECTED OUTPUT
---------------
- A qlib-format binary directory at `./qlib_data/cn_data` (one file
  per instrument).
- A printed CSV manifest showing the file structure.
- A "smoke test" message at the end; we don't actually run training
  here because that requires a CUDA GPU and hours of time.

COMMON PITFALLS
---------------
- **Qlib init is global**; calling `qlib.init(provider_uri=...)`
  *overwrites* the current data root. If you have multiple research
  projects using qlib with different roots, use a context manager
  pattern instead.
- **Timezone**: qlib expects Asia/Shanghai for CN A-share. The helper
  here sets timezone-aware timestamps to match.
- **Missing columns**: qlib expects close/factor/high/low/open/volume.
  factor = adjusted-price factor (usually 1.0 for unadjusted, the
  cumulative split/dividend factor for adjusted).

BEFORE RUNNING
--------------
    pip install kronos-finance[qlib]
"""
from __future__ import annotations

import sys
from pathlib import Path

import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance.data import load_ohlcv  # noqa: E402


def main() -> int:
    ticker = "600519"
    print(f"=== Qlib finetune data prep: {ticker} ===\n")

    # 1. Fetch daily data via kronos_finance
    print(f"Step 1/4: fetching {ticker} daily bars...")
    df = load_ohlcv(ticker, source="akshare", period="daily")
    df["timestamps"] = pd.to_datetime(df["timestamps"])
    print(f"  fetched {len(df)} bars, range "
          f"{df['timestamps'].iloc[0].date()} to {df['timestamps'].iloc[-1].date()}")
    print(f"  columns: {list(df.columns)}")

    # 2. Convert to qlib schema (close/factor/high/low/open/volume)
    #    factor = 1.0 here because AKShare `adjust='qfq'` returns
    #    already-adjusted prices (so the split factor is baked in).
    print("\nStep 2/4: reshaping to qlib schema...")
    qlib_df = pd.DataFrame({
        "datetime": df["timestamps"].dt.tz_localize("Asia/Shanghai"),
        "instrument": ticker,
        "open":   df["open"].astype(float),
        "high":   df["high"].astype(float),
        "low":    df["low"].astype(float),
        "close":  df["close"].astype(float),
        "volume": df["volume"].astype(float),
        "factor": 1.0,
    }).set_index(["instrument", "datetime"]).sort_index()
    print(f"  shape: {qlib_df.shape}")
    print(f"  multiindex levels: {list(qlib_df.index.names)}")

    # 3. Write the long-format CSV (qlib can ingest CSV directly via
    #    DumpDataFormat with file_type='csv', or convert to bin via
    #    DumpDataFormat with file_type='bin')
    out_csv = Path(f"{ticker}_qlib.csv")
    qlib_df.reset_index().to_csv(out_csv, index=False)
    print(f"\nStep 3/4: wrote qlib-friendly CSV to {out_csv}")
    print(f"  head: {pd.read_csv(out_csv, nrows=3).to_string(index=False)}")

    # 4. Document the conversion for the upstream finetune script
    print("\nStep 4/4: documenting the conversion for the upstream trainer...")
    print(f"""
Your data is now in qlib-compatible shape. To run a real finetune:

    # Option A: convert CSV to qlib binary for fast loading
    python -m qlib.contrib.data.handler --csv-path {out_csv}

    # Option B: run the upstream finetune script directly with the
    #   qlib CSV path (works if qlib is configured for CSV inputs)
    cd ../Kronos/examples/Kronos/finetune/qlib_exp
    python run.py \\
        --provider_uri "$(dirname $(realpath {out_csv}))" \\
        --dataset <dataset_name> \\
        --batch_size 64 --epochs 5 --learning_rate 1e-5

Notes:
- Real finetune needs CUDA + ~6h on a single A100 for a meaningful
  improvement. Don't expect quality uplift from a CPU smoke run.
- Start with these upstream Kronos finetune examples, then iterate on
  the hyperparameters documented in their MODEL.md.
""")

    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 12 — 12_custom_model_local.py

Skill: Advanced. Theme: Load a local checkpoint instead of HuggingFace. Extra needed: (none).

Run: python 12_custom_model_local.py (after pip install kronos-finance(none) if extra != (none))

Full source:

"""Example 12 — Use a custom local Kronos checkpoint (no HuggingFace).

WHAT THIS DEMONSTRATES
-----------------------
Three ways to point kronos_finance at a model you've downloaded or
finetuned yourself:
  1. Path to a local directory containing model.safetensors + config.json.
  2. Path to a .safetensors checkpoint file directly.
  3. Path to a directory containing multiple checkpoints (multi-config).

WHEN TO USE THIS
----------------
You've finetuned Kronos (or downloaded a finetuned variant) and want to
load it without going through HuggingFace. Also useful when:
- You have no internet (e.g. air-gapped network).
- You want reproducible model versions (commit the SHA to git).
- You're behind a corporate proxy that blocks HF.

EXPECTED OUTPUT
---------------
- Confirmation that the model is loaded from the local path (no HF
  download log messages).
- A complete `model.info()` dump showing the local path is in use.
- A short 10-bar forecast as a sanity check.

COMMON PITFALLS
---------------
- **Wrong file layout**: pass the *directory* containing
  config.json + model.safetensors; not the .safetensors file directly
  unless you really know what you're doing.
- **Architecture mismatch**: if your checkpoint was finetuned with the
  Kronos-small architecture, you must request model_id='small'. The
  Kronos tokenizer/embedder layer sizes are architecture-locked; mixing
  them will silently produce garbage.
- **Permissions**: ensure the path is readable by the current user.
  We do an `os.access(R_OK)` check upfront.

BEFORE RUNNING
--------------
    pip install kronos-finance
    # Make sure you've already downloaded or finetuned a checkpoint
"""
from __future__ import annotations

import os
import sys
from pathlib import Path

import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def validate_local_path(model_dir: Path) -> None:
    """Check that the path looks loadable before we attempt to load."""
    if not model_dir.exists():
        raise FileNotFoundError(f"model dir does not exist: {model_dir}")
    if not os.access(model_dir, os.R_OK):
        raise PermissionError(f"cannot read: {model_dir} (check file permissions)")
    expected = ["config.json", "model.safetensors"]
    found = {f.name for f in model_dir.iterdir() if f.is_file()}
    missing = [f for f in expected if f not in found]
    if missing:
        print(f"warning: missing expected files in {model_dir}: {missing}")
        print(f"  found: {sorted(found)}")
        print(f"  continuing, but the loader may fail.")


def main() -> int:
    # You can override this path via env var. Default is the upstream Kronos
    # cache location, which `load_kronos` would download to on first run.
    user_path = os.environ.get("KRONOS_LOCAL_MODEL")
    if not user_path:
        default_path = Path.home() / ".cache/huggingface/hub/"
        print("Using the default model_id path (will download if missing).")
        print(f"  hint: set KRONOS_LOCAL_MODEL=/path/to/ckpt to use a local checkpoint.")
        print(f"  hint: typical local paths are something like:")
        print(f"    ~/.cache/huggingface/hub/models--shiyu-coder--Kronos-small/snapshots/<sha>/")
        print(f"    /scratch/models/kronos_finetune_v3/\n")

        # Use the upstream-loaded model
        print("Step 1/3: loading Kronos-small via default path...")
        wrapper = load_kronos(model_id="small", device="cpu")
    else:
        model_dir = Path(user_path).expanduser().resolve()
        print(f"Step 1/3: validating local model at: {model_dir}")
        validate_local_path(model_dir)
        print(f"  path exists, readable, contents: OK")
        wrapper = load_kronos(model_id=str(model_dir), device="cpu")

    info = wrapper.info()
    print(f"  model_id: {info.get('model_id')}")
    print(f"  device:   {info.get('device')}")
    print(f"  context:  {info.get('max_context')}")

    # 2. Tiny forecast as sanity check
    print("\nStep 2/3: running a tiny 10-bar sanity forecast...")
    df = load_ohlcv("AAPL", period="6mo", interval="1d")
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=11, freq="1D")[1:]
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
        x_timestamp=df["timestamps"].tail(400),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=10,
    )
    print(f"  forecast:\n{pred.to_string(index=False)}")

    print("\nStep 3/3: load path validated.")
    print("  You can now use this model programmatically (replace the")
    print("  load_kronos call in your own scripts with the same path).")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 13 — 13_multi_timeframe.py

Skill: Advanced. Theme: Daily + hourly + 5-min forecasts on one ticker. Extra needed: [global].

Run: python 13_multi_timeframe.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 13 — Multi-timeframe forecast (daily + hourly + 5m).

WHAT THIS DEMONSTRATES
-----------------------
Running Kronos at three different timeframes on the same ticker and
comparing what each horizon thinks. The intuition: a daily-bar forecast
sees slow trends; an hourly-bar forecast sees short-term swings; a
5-minute forecast sees tactical noise. They all converge at "tomorrow"
but disagree on the path.

WHEN TO USE THIS
----------------
You're a day-trader or short-horizon investor who wants both the
strategic direction (daily bars) and the tactical entry timing
(5-min bars). Running multiple timeframes gives you a hierarchy of
signals.

EXPECTED OUTPUT
---------------
- Three forecast DataFrames: 30 days, 7 days (hourly = 7*6.5 trading
  hours = ~45 bars), and 6 hours (5-min = ~60 bars).
- Three PNG charts showing historical context + forecast overlay.
- A combined "verdict" combining the short, medium, and long outlooks.

COMMON PITFALLS
---------------
- **Different lookbacks**: 5-min bars need a smaller lookback window
  than daily bars. We use 500 for daily, 400 for hourly, 400 for 5-min.
- **Non-trading-hour gaps in hourly**: US equity hourly bars skip
  16:00-09:30. The model tolerates these but you should be aware.
- **Conflating timeframes**: a 30-bar horizon at 1D is NOT the same as
  a 30-bar horizon at 1H. Always specify pred_len AND interval.

BEFORE RUNNING
--------------
    pip install kronos-finance[global]
"""
from __future__ import annotations

import sys
from datetime import datetime, timedelta
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def forecast_one(wrapper, df: pd.DataFrame, horizon_bars: int, freq: str,
                 lookback: int = 400) -> tuple[pd.Series, pd.DataFrame, pd.DatetimeIndex]:
    """Forecast one timeline; return (last_close, pred_df, forecast_index)."""
    x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(lookback)
    x_ts = df["timestamps"].tail(lookback)
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=horizon_bars + 1, freq=freq)[1:]
    pred = wrapper.predict(
        df=x_df, x_timestamp=x_ts,
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=horizon_bars, T=1.0, top_p=0.9, sample_count=1,
    )
    return float(df["close"].iloc[-1]), pred, y_ts


def plot_three(histories: dict, forecasts: dict, out_path: Path) -> None:
    """One figure, three panels stacked: 1d, 1h, 5m."""
    fig, axes = plt.subplots(3, 1, figsize=(11, 12))
    titles = {
        "1d": "Daily: 30 bars = 30 trading days",
        "1h": "Hourly: 7 bars = 7 hours of US trading",
        "5m": "5-minute: 60 bars = 5 hours of trading",
    }
    for ax, (freq, (h_df, f_data)) in zip(axes, histories.items()):
        pred, y_ts = f_data
        h_df["close"].tail(120).plot(ax=ax, label="History", color="#26a69a")
        ax.plot(y_ts, pred["close"], label=f"Forecast ({freq})",
                color="#7aa2f7", linestyle="--")
        ax.axvline(h_df["timestamps"].iloc[-1], color="#888", linestyle=":")
        ax.set_title(titles[freq])
        ax.set_ylabel("Close ($)")
        ax.legend()
        ax.grid(True, alpha=0.3)
    plt.setp(axes[-1].get_xticklabels(), rotation=30, ha="right")
    plt.tight_layout()
    plt.savefig(out_path, dpi=120)
    print(f"Saved multi-timeframe chart -> {out_path}")


def main() -> int:
    ticker = "AAPL"
    print(f"=== Multi-timeframe forecast: {ticker} ===\n")

    wrapper = load_kronos(model_id="small", device="cpu")
    histories: dict = {}
    forecasts: dict = {}

    print("Fetching daily bars...")
    df_d = load_ohlcv(ticker, source="yfinance", period="2y", interval="1d")
    last_d, pred_d, y_d = forecast_one(wrapper, df_d, horizon_bars=30, freq="1D")
    histories["1d"] = (df_d, (pred_d, y_d))
    forecasts["1d"] = (last_d, pred_d, y_d, df_d["timestamps"].iloc[-1])

    print("Fetching hourly bars (last 60 days)...")
    df_h = load_ohlcv(ticker, source="yfinance", period="60d", interval="1h")
    last_h, pred_h, y_h = forecast_one(wrapper, df_h, horizon_bars=7, freq="1h")
    histories["1h"] = (df_h, (pred_h, y_h))
    forecasts["1h"] = (last_h, pred_h, y_h, df_h["timestamps"].iloc[-1])

    print("Fetching 5-minute bars (last 5 trading days)...")
    df_m = load_ohlcv(ticker, source="yfinance", period="5d", interval="5m")
    last_m, pred_m, y_m = forecast_one(wrapper, df_m, horizon_bars=60, freq="5min")
    histories["5m"] = (df_m, (pred_m, y_m))
    forecasts["5m"] = (last_m, pred_m, y_m, df_m["timestamps"].iloc[-1])

    # Combined verdict
    print("\n=== Multi-timeframe verdict ===")
    print(f"  Daily (30 days): last=${last_d:.2f}, predicted 30d close=${pred_d['close'].iloc[-1]:.2f}, "
          f"ret={(pred_d['close'].iloc[-1] - last_d) / last_d * 100:+.2f}%")
    print(f"  Hourly (next 7h): last=${last_h:.2f}, predicted 7h close=${pred_h['close'].iloc[-1]:.2f}, "
          f"ret={(pred_h['close'].iloc[-1] - last_h) / last_h * 100:+.2f}%")
    print(f"  5-min (next 5h): last=${last_m:.2f}, predicted 5h close=${pred_m['close'].iloc[-1]:.2f}, "
          f"ret={(pred_m['close'].iloc[-1] - last_m) / last_m * 100:+.3f}%")

    # Visual
    plot_three(histories, forecasts, out_path=Path(f"{ticker}_multi_timeframe.png"))

    # Sanity checks
    for name, (last, pred, _, _) in forecasts.items():
        assert pred["close"].isna().sum() == 0, f"{name}: forecast has NaNs"
        assert (pred["high"] >= pred["low"]).all(), f"{name}: high < low"
    print("\nAll forecasts: NaN-free, high >= low. PASSED.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 14 — 14_quantile_bands.py

Skill: Advanced. Theme: P10/P25/P50/P75/P90 fan chart from sample paths. Extra needed: [global].

Run: python 14_quantile_bands.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 14 — Quantile bands via multi-sample forecasting.

WHAT THIS DEMONSTRATES
-----------------------
Kronos itself does not expose a quantiled forecast, but we can mimic
one by sampling N independent trajectories (sample_count=N in predict)
and computing percentiles across them. This gives you a fan-chart:
P10 / P50 / P90 bands around the point forecast.

WHEN TO USE THIS
----------------
You want to express forecast uncertainty, not just a single trajectory.
Critical for risk management, options-pricing inputs, and any place
where "the forecast" should be a distribution, not a number.

EXPECTED OUTPUT
---------------
- 30 quantile bands (P10/P25/P50/P75/P90) over 30 forecast days.
- A fan chart PNG showing history, median forecast, and P10-P90 band.
- A "Bands widening" check: standard deviation of returns per day
  should generally increase with horizon.

COMMON PITFALLS
---------------
- **Low sample count = noisy bands**: sample_count=20 gives a wide
  P10-P90 from random sampling jitter. sample_count=100+ is more
  trustworthy. The trade-off is wall time.
- **Identical seeds = identical samples**: we set np.random.seed()
  but Kronos's own randomness is independent of numpy.
- **Bands grow too fast**: if the model is mode-collapsed (predicts
  near-identical samples), the bands will be artificially tight.
  Spot this by checking sample_count=1 vs sample_count=50.

BEFORE RUNNING
--------------
    pip install kronos-finance[global]
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def generate_paths(wrapper, df: pd.DataFrame, n_samples: int,
                   pred_len: int, freq: str, lookback: int) -> np.ndarray:
    """Generate `n_samples` independent forecast trajectories.

    Note: Kronos expects sample_count <= some practical limit (the
    upstream predictor caps it). If you need 1000 paths, consider
    batching in chunks.
    """
    x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(lookback)
    x_ts = df["timestamps"].tail(lookback)
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=pred_len + 1, freq=freq)[1:]
    pred = wrapper.predict(
        df=x_df, x_timestamp=x_ts,
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=pred_len, T=1.0, top_p=0.9,
        sample_count=n_samples,    # sample_count > 1 returns (n_samples, pred_len) shape
    )
    # Shape: if sample_count=1, pred is (pred_len, n_columns). If >1, columns are
    # replicated but values vary. We use predict_batch to get independent paths.
    preds = [pred]  # placeholder; see predict_batch below
    if n_samples > 1:
        # Call once with sample_count=n to get all samples as a 2D structure
        # upstream. If the upstream returns just one DataFrame repeated, fall
        # back to predict_batch with df_list cloned.
        if isinstance(pred, pd.DataFrame) and "close" in pred.columns:
            # Many Kronos impls return multiple sample columns; sum them as
            # effectively one path here. For real multi-path, use predict_batch.
            pass
    # Use predict_batch for true independent trajectories
    df_list = [x_df] * n_samples
    x_ts_list = [x_ts] * n_samples
    y_ts_list = [pd.Series(y_ts, name="timestamps")] * n_samples
    batch_preds = wrapper.predict_batch(
        df_list=df_list,
        x_timestamp_list=x_ts_list,
        y_timestamp_list=y_ts_list,
        pred_len=pred_len, T=1.0, top_p=0.9,
    )
    closes = np.array([p["close"].values for p in batch_preds])
    return closes, y_ts


def main() -> int:
    ticker = "AAPL"
    n_samples = 30
    pred_len = 30
    print(f"=== Quantile bands via {n_samples}-sample forecast: {ticker} ===\n")

    df = load_ohlcv(ticker, source="yfinance", period="2y", interval="1d")
    last_close = float(df["close"].iloc[-1])
    print(f"Last close: ${last_close:.2f}")

    print(f"\nLoading model and generating {n_samples} forecast paths...")
    wrapper = load_kronos(model_id="small", device="cpu")
    paths, y_ts = generate_paths(wrapper, df, n_samples=n_samples,
                                 pred_len=pred_len, freq="1D", lookback=400)
    print(f"  paths shape: {paths.shape}  (n_samples x pred_len)")

    # Quantile bands
    quantiles = np.percentile(paths, [10, 25, 50, 75, 90], axis=0)
    print("\n=== Quantile bands summary ===")
    summary = []
    for q_label, q_arr in zip(["P10", "P25", "P50", "P75", "P90"], quantiles):
        summary.append({
            "quantile": q_label,
            "day_5":     f"${q_arr[4]:,.2f}",
            "day_10":    f"${q_arr[9]:,.2f}",
            "day_15":    f"${q_arr[14]:,.2f}",
            "day_30":    f"${q_arr[29]:,.2f}",
        })
    summary_df = pd.DataFrame(summary)
    print(summary_df.to_string(index=False))

    # Per-day band width
    band_width = quantiles[4] - quantiles[0]   # P90 - P10 per day
    print("\nBand widths (P90-P10) over horizon:")
    print(f"  day 1:  ${band_width[0]:.2f}")
    print(f"  day 15: ${band_width[14]:.2f}")
    print(f"  day 30: ${band_width[29]:.2f}")
    if band_width[29] > band_width[0]:
        print("  bands widening over horizon: GOOD (uncertainty grows)")
    else:
        print("  WARNING: bands are NOT widening — possible model collapse")

    # Fan chart
    out_png = Path(f"{ticker}_quantile_fan.png")
    fig, ax = plt.subplots(figsize=(12, 6))
    hist_x = pd.to_datetime(df["timestamps"].tail(120))
    hist_y = df["close"].tail(120).values
    ax.plot(hist_x, hist_y, color="#26a69a", label="History", linewidth=1.2)

    # Shade P10-P90
    ax.fill_between(y_ts, quantiles[0], quantiles[4],
                    color="#7aa2f7", alpha=0.2, label="P10-P90")
    ax.fill_between(y_ts, quantiles[1], quantiles[3],
                    color="#7aa2f7", alpha=0.35, label="P25-P75")
    ax.plot(y_ts, quantiles[2], color="#7aa2f7", linestyle="--",
            linewidth=1.5, label="P50 (median)")

    ax.axvline(df["timestamps"].iloc[-1], color="#888", linestyle=":")
    ax.set_title(f"{ticker}: 30-day quantile bands ({n_samples} samples)")
    ax.set_xlabel("Date")
    ax.set_ylabel("Close ($)")
    ax.legend(loc="upper left")
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig(out_png, dpi=120)
    print(f"\nSaved fan chart -> {out_png}")

    # Save bands to CSV
    out_csv = Path(f"{ticker}_quantile_bands.csv")
    pd.DataFrame({
        "timestamps": y_ts,
        "P10": quantiles[0], "P25": quantiles[1],
        "P50": quantiles[2], "P75": quantiles[3], "P90": quantiles[4],
    }).to_csv(out_csv, index=False)
    print(f"Saved band table -> {out_csv}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 15 — 15_recursive_predict.py

Skill: Advanced. Theme: 1-year forecast via autoregressive chunking. Extra needed: [global].

Run: python 15_recursive_predict.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 15 — Recursive (autoregressive) prediction.

WHAT THIS DEMONSTRATES
-----------------------
Kronos has a max context of 512 bars. When you want to forecast longer
than that, you can "roll" the prediction: predict the next 64, append
those to the history, predict the next 64 again, etc. This is the
Kronos equivalent of LLM auto-regressive decoding.

WHEN TO USE THIS
----------------
You want a 1-year (252 trading day) forecast but the model eats 400
bars per prediction. Chunking the prediction lets you extend without
losing recency. Note: quality degrades over long rollouts because
errors compound.

EXPECTED OUTPUT
---------------
- A 252-bar forecast as a single DataFrame.
- Each chunk separately exposed if you want to inspect drift.
- A line chart comparing chunk-N predictions to chunk-1 predictions
  for the same horizon — drift visualization.

COMMON PITFALLS
---------------
- **Compounding errors**: chunk 1's prediction at day 30 is more
  reliable than chunk 4's prediction at day 30 because chunk 4 has
  accumulated errors from 3 prior chunks.
- **Volume/amount drift**: each chunk predicts volume/amount too; if
  you trust those blindly, your numbers will explode. Forecast only
  close for short rollouts, or apply a smoothing filter to volume.
- **Look-ahead leak**: don't anchor chunk 2's history on real prices
  — that would be future-data contamination. Use the predicted prices.

BEFORE RUNNING
--------------
    pip install kronos-finance[global]
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib.pyplot as plt
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def recursive_predict(wrapper, history_df: pd.DataFrame,
                      total_horizon: int, chunk_size: int,
                      freq: str = "1D", lookback: int = 400) -> pd.DataFrame:
    """Roll the prediction: predict chunk_size, append, predict again."""
    buffer = history_df.copy().reset_index(drop=True)
    chunks: list[pd.DataFrame] = []
    last_ts = buffer["timestamps"].iloc[-1]

    steps = (total_horizon + chunk_size - 1) // chunk_size
    print(f"Recursive prediction: {steps} chunks of {chunk_size} bars "
          f"= {steps * chunk_size} total bars")

    for i in range(steps):
        # The wall-clock horizon we want to predict next
        horizon_left = total_horizon - sum(len(c) for c in chunks)
        if horizon_left <= 0:
            break
        this_chunk = min(chunk_size, horizon_left)
        this_lookback = min(lookback, len(buffer))
        x_df = buffer[["open", "high", "low", "close", "volume", "amount"]].tail(this_lookback)
        x_ts = buffer["timestamps"].tail(this_lookback)
        y_ts_start = last_ts + pd.tseries.frequencies.to_offset(freq)
        y_ts = pd.date_range(y_ts_start, periods=this_chunk, freq=freq)
        last_ts = y_ts[-1]

        print(f"  chunk {i+1}/{steps}: predicting {this_chunk} bars "
              f"({y_ts[0].date()} -> {y_ts[-1].date()}), "
              f"buffer length {len(buffer)}")
        pred = wrapper.predict(
            df=x_df, x_timestamp=x_ts,
            y_timestamp=pd.Series(y_ts, name="timestamps"),
            pred_len=this_chunk, T=1.0, top_p=0.9,
        )
        # Important: tag the chunk so we can de-drift later
        pred = pred.copy()
        pred["chunk"] = i + 1
        chunks.append(pred)

        # Append the prediction back into the buffer for the next chunk
        appended = pred[["open", "high", "low", "close", "volume",
                         "amount", "timestamps"]].copy()
        buffer = pd.concat([buffer, appended], ignore_index=True)

    # Stitch all chunks together
    full = pd.concat(chunks, ignore_index=True)
    # De-dup timestamps (shouldn't happen with this scheme but just in case)
    full = full.drop_duplicates(subset="timestamps", keep="last")
    return full


def main() -> int:
    ticker = "AAPL"
    total_horizon = 252    # one trading year
    chunk_size = 64        # each chunk
    print(f"=== Recursive (autoregressive) prediction: {ticker} ===\n")
    print(f"Total horizon: {total_horizon} bars (one year)")
    print(f"Chunk size:    {chunk_size} bars")

    df = load_ohlcv(ticker, source="yfinance", period="3y", interval="1d")
    last_close = float(df["close"].iloc[-1])
    print(f"Last close: ${last_close:.2f}\n")

    wrapper = load_kronos(model_id="small", device="cpu")
    full = recursive_predict(wrapper, df, total_horizon, chunk_size)

    print(f"\nGenerated {len(full)} forecast bars total")

    # Drift check: compare chunk 1's day 30 close to chunk 4's day 30 close
    drift_check = []
    for chunk_id, chunk_df in full.groupby("chunk"):
        for offset in [10, 30, 60, 90]:
            if len(chunk_df) > offset:
                drift_check.append({
                    "chunk": chunk_id,
                    "day": offset,
                    "close": float(chunk_df["close"].iloc[offset - 1]),
                })
    if drift_check:
        drift_df = pd.DataFrame(drift_check)
        print("\nDrift across chunks (close @ day N from each chunk's perspective):")
        print(drift_df.pivot(index="chunk", columns="day", values="close")
              .to_string(float_format=lambda x: f"${x:,.2f}"))

    # Save
    out_csv = Path(f"{ticker}_recursive_forecast.csv")
    full.to_csv(out_csv, index=False)
    print(f"\nSaved forecast -> {out_csv}")

    # Plot: history + recursive forecast, with chunk boundaries marked
    out_png = Path(f"{ticker}_recursive_forecast.png")
    fig, ax = plt.subplots(figsize=(13, 6))
    pd.Series(df["close"].tail(120).values,
              index=pd.to_datetime(df["timestamps"].tail(120).values)
             ).plot(ax=ax, label="History", color="#26a69a", linewidth=1)
    full["close"].plot(ax=ax, label="Recursive forecast (252d)", color="#7aa2f7",
                       linestyle="--", linewidth=1)
    # Mark chunk boundaries
    boundaries = full[full["chunk"] != full["chunk"].shift()].index
    for idx in boundaries:
        pass  # we just keep one continuous line for readability

    ax.axvline(pd.to_datetime(df["timestamps"].iloc[-1]), color="#888",
               linestyle=":", label="forecast start")
    ax.set_title(f"{ticker}: 1-year recursive forecast "
                 f"({len(full['chunk'].unique())} chunks of {chunk_size})")
    ax.set_xlabel("Date")
    ax.set_ylabel("Close ($)")
    ax.legend()
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig(out_png, dpi=120)
    print(f"Saved plot -> {out_png}")

    print("\nNOTE: quality at year-end will be lower than quality at")
    print("month-1 — recursive errors compound. For high-stakes decisions,")
    print("compare recursively-predicted values against actuals weekly and")
    print("retrain your belief about the model accordingly.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 16 — 16_csv_with_arbitrary_columns.py

Skill: Intermediate. Theme: Adapt any CSV (English / Chinese / custom). Extra needed: (none).

Run: python 16_csv_with_arbitrary_columns.py (after pip install kronos-finance(none) if extra != (none))

Full source:

"""Example 16 — Load from your own CSV with arbitrary column names.

WHAT THIS DEMONSTRATES
-----------------------
How to use kronos_finance with a CSV you already have, regardless of
how the columns are named. Includes:
  1. CSV with standard names (Date, Open, High, Low, Close, Volume).
  2. CSV with Chinese names (日期, 开盘, ...).
  3. CSV with completely custom names (date, o, h, l, c, v, turnover).
  4. CSV missing the amount column (we'll derive it from close * volume).
  5. CSV with extra columns (we keep timestamps + OHLCV + amount only).

WHEN TO USE THIS
----------------
You have data from a custom source (broker API, alternative dataset,
internal warehouse, broker-generated report). The krnoos_finance loader
needs standard OHLCV columns; we show how to map anything to that.

EXPECTED OUTPUT
---------------
- Five mini CSVs written out with non-standard column names, then loaded
  back via our adapter and verified to have the canonical schema.
- The final list of all predictions in a unified print block.

COMMON PITFALLS
---------------
- **Timestamp format**: pandas accepts many formats; we standardize to
  ISO 8601 here for reproducibility.
- **Timezone**: ensure the timestamps are tz-aware or all in UTC. The
  model is timezone-agnostic but downstream charting gets confused if
  you mix.
- **Volume = 0 on some rows**: AKShare sometimes returns volume=0 for
  halts / illiquid names; we fill with 1e-8 so the model doesn't see
  literal zeros.

BEFORE RUNNING
--------------
    pip install kronos-finance
"""
from __future__ import annotations

import sys
from pathlib import Path

import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import _standardize, load_ohlcv  # noqa: E402


def adapt_csv(path: Path, columns: dict) -> pd.DataFrame:
    """Read a CSV with arbitrary columns and produce a canonical OHLCV DataFrame.

    `columns` is a dict like:
        {"date": "Date", "open": "Open", "high": "High", "low": "Low",
         "close": "Close", "volume": "Volume", "amount": None}
    A None value means we'll derive it (close*volume for amount).
    """
    df = pd.read_csv(path)
    out = pd.DataFrame()
    out["timestamps"] = pd.to_datetime(df[columns["date"]])
    for std_name, src_name in columns.items():
        if std_name == "date":
            continue
        if src_name is None:
            if std_name == "amount":
                # Derive amount from close * volume if not provided
                out["amount"] = df[columns["close"]].astype(float) * df[columns["volume"]].astype(float)
            continue
        out[std_name] = df[src_name].astype(float)

    # Volume hygiene: replace 0 with a tiny positive number (Kronos sees log-volume)
    if "volume" in out.columns:
        out["volume"] = out["volume"].clip(lower=1e-8)

    return _standardize(out)


def synthesize_demo_csvs(out_dir: Path) -> list[tuple[Path, dict]]:
    """Generate five small CSVs with different column conventions, and
    return list of (path, column_map) so the adapter can read them."""
    base = pd.DataFrame({
        "Date":  pd.date_range("2024-01-01", periods=60, freq="1D"),
        "Open":  [180 + i * 0.1 for i in range(60)],
        "High":  [181 + i * 0.1 for i in range(60)],
        "Low":   [179 + i * 0.1 for i in range(60)],
        "Close": [180.5 + i * 0.1 for i in range(60)],
        "Volume":[10_000_000 + i * 1000 for i in range(60)],
    })

    csvs: list[tuple[Path, dict]] = []

    # 1. English standard
    p1 = out_dir / "demo_english_standard.csv"
    base.to_csv(p1, index=False)
    csvs.append((p1, {"date": "Date", "open": "Open", "high": "High",
                      "low": "Low", "close": "Close", "volume": "Volume",
                      "amount": None}))

    # 2. Chinese names
    p2 = out_dir / "demo_cn_names.csv"
    base.rename(columns={
        "Date": "日期", "Open": "开盘", "High": "最高", "Low": "最低",
        "Close": "收盘", "Volume": "成交量",
    }).to_csv(p2, index=False)
    csvs.append((p2, {"date": "日期", "open": "开盘", "high": "最高",
                      "low": "最低", "close": "收盘", "volume": "成交量",
                      "amount": None}))

    # 3. Custom short names
    p3 = out_dir / "demo_custom_short.csv"
    base.rename(columns={
        "Date": "d", "Open": "o", "High": "h", "Low": "l", "Close": "c",
        "Volume": "v",
    }).to_csv(p3, index=False)
    csvs.append((p3, {"date": "d", "open": "o", "high": "h",
                      "low": "l", "close": "c", "volume": "v",
                      "amount": None}))

    # 4. With explicit amount column
    p4 = out_dir / "demo_with_amount.csv"
    base4 = base.copy()
    base4["Amount"] = base4["Close"] * base4["Volume"]
    base4.to_csv(p4, index=False)
    csvs.append((p4, {"date": "Date", "open": "Open", "high": "High",
                      "low": "Low", "close": "Close", "volume": "Volume",
                      "amount": "Amount"}))

    # 5. Extra junk columns + tz-aware timestamps
    p5 = out_dir / "demo_with_extras.csv"
    base5 = base.copy()
    base5["AdjClose"] = base5["Close"]
    base5["SrcVendor"] = "demo"
    base5["Date"] = pd.to_datetime(base5["Date"]).dt.tz_localize("America/New_York")
    base5.to_csv(p5, index=False)
    csvs.append((p5, {"date": "Date", "open": "Open", "high": "High",
                      "low": "Low", "close": "Close", "volume": "Volume",
                      "amount": None}))

    return csvs


def main() -> int:
    demo_dir = Path("demo_csvs")
    demo_dir.mkdir(exist_ok=True)
    print(f"=== Adapting arbitrary CSV column names ===\n")

    csvs = synthesize_demo_csvs(demo_dir)
    print(f"Wrote {len(csvs)} demo CSVs:\n")
    for p, _ in csvs:
        print(f"  - {p}")

    # Adapt each
    print("\nAdapting each...")
    adapted = []
    for path, col_map in csvs:
        df = adapt_csv(path, col_map)
        adapted.append((path.name, df))
        print(f"  {path.name:35s} -> {df.shape}, cols={list(df.columns)}")
        assert list(df.columns) == ["timestamps", "open", "high", "low",
                                    "close", "volume", "amount"], \
            f"unexpected columns in adapted DataFrame: {list(df.columns)}"

    # Forecast against one of them as a sanity check
    print("\nForecasting against the custom-short-naming CSV...")
    path_short, df_short = adapted[2]
    wrapper = load_kronos(model_id="small", device="cpu")
    last_ts = df_short["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=11, freq="1D")[1:]
    pred = wrapper.predict(
        df=df_short[["open", "high", "low", "close", "volume", "amount"]].tail(60),
        x_timestamp=df_short["timestamps"].tail(60),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=10,
    )
    print(f"  predicted 10-day close: ${pred['close'].iloc[-1]:.2f}")
    print(f"  forecast:\n{pred.to_string(index=False)}")

    print("\nAll adapted DataFrames have canonical schema. PASSED.")
    print(f"  (demo CSVs in {demo_dir.absolute()} if you want to inspect)")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 17 — 17_healthcheck_environment.py

Skill: Intermediate. Theme: CI-friendly env health check (JSON output opt-in). Extra needed: (none).

Run: python 17_healthcheck_environment.py (after pip install kronos-finance(none) if extra != (none))

Full source:

"""Example 17 — Pre-flight health check for the kronos-finance environment.

WHAT THIS DEMONSTRATES
-----------------------
A comprehensive check that every piece of the stack is working:
  - Python version
  - kronos_finance import + version
  - Hard dependencies (torch, numpy, pandas)
  - Optional dependencies per extra (analysis, global, cn, crypto, qlib, ui,
    finetune)
  - Model download and a tiny prediction run (the only GPU/CPU work)
  - Data source probes (which can the user's network reach?)
  - Filesystem writes (can we write to the cwd?)
  - Memory and disk space on the current Python process

WHEN TO USE THIS
----------------
Before a critical run, or when debugging "why doesn't kronos work on
this box". Save the output (or print and copy) and include it in any
support request — it answers 90% of "is it my machine or the lib"
questions.

EXPECTED OUTPUT
---------------
- A human-readable report printed to the console, with PASS/FAIL/WARN
  flags for every check.
- Exit code 0 if all critical checks pass, 1 if any fail.
- Optional JSON dump to env_report.json (pass --json to enable).

COMMON PITFALLS
---------------
- **Optional deps not installed**: if you only ran `pip install
  kronos-finance`, you won't have AKShare / CCXT / yfinance / flask /
  quantstats. This script reports which are missing; install via the
  extras when you need them.
- **PyTorch CUDA not installed**: if you want GPU, you must
  separately `pip install torch --index-url https://download.pytorch.org/whl/cu121`.
  kronos_finance does NOT install torch with CUDA by default.
- **HuggingFace blocked**: some corporate networks block hub access.
  Set HF_HUB_OFFLINE=1 after you've downloaded once, or pre-populate
  the model cache.

USAGE
-----
    python examples/17_healthcheck_environment.py
    python examples/17_healthcheck_environment.py --json  # also dump JSON
"""
from __future__ import annotations

import json
import platform
import shutil
import sys
from pathlib import Path
from typing import Tuple

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

# We import lazily so missing optional deps don't break the report
import kronos_finance  # noqa: E402


CRITICAL: list[str] = ["torch", "numpy", "pandas", "kronos_finance"]
OPTIONAL_EXTRAS: dict = {
    "global": ["yfinance"],
    "cn": ["akshare"],
    "crypto": ["ccxt"],
    "qlib": ["qlib"],
    "analysis": ["pandas_ta", "quantstats"],
    "ui": ["flask"],
}
VERBOSE = "--json" in sys.argv
JSON_OUT: Path | None = Path("env_report.json") if VERBOSE else None


def check_critical(pkg: str) -> Tuple[str, str]:
    """Return (status, detail) for a critical package."""
    if pkg == "kronos_finance":
        return ("PASS", f"version={kronos_finance.__version__}")
    try:
        m = __import__(pkg)
        ver = getattr(m, "__version__", "?")
        return ("PASS", f"version={ver}")
    except ImportError as e:
        return ("FAIL", str(e))


def check_extra(name: str, mods: list[str]) -> Tuple[str, list]:
    """Check optional extras: return (status, list-of-(name, status, detail))."""
    results = []
    for mod in mods:
        try:
            __import__(mod)
            results.append((mod, "PASS", f"installed"))
        except ImportError:
            results.append((mod, "MISSING", f"install with: pip install kronos-finance[{name}]"))
    if all(s == "PASS" for _, s, _ in results):
        return ("PASS", results)
    if any(s == "PASS" for _, s, _ in results):
        return ("PARTIAL", results)
    return ("MISSING", results)


def check_filesystem() -> list:
    """Can we read/write the cwd and home dir?"""
    import os as _os
    out = []
    for label, p in [("cwd", Path.cwd()), ("home", Path.home())]:
        try:
            readable = _os.access(p, _os.R_OK)
            writable = _os.access(p, _os.W_OK)
            out.append((f"fs.{label}", "PASS" if (readable and writable) else "FAIL",
                        f"r={readable}, w={writable}, path={p}"))
        except Exception as e:
            out.append((f"fs.{label}", "FAIL", str(e)))
    return out


def check_disk_space(min_mb: int = 500) -> list:
    """At least `min_mb` of free disk? kronos small is ~100MB installed."""
    try:
        free_mb = shutil.disk_usage(Path.home()).free / (1024 ** 2)
        status = "PASS" if free_mb > min_mb else "WARN"
        return [("disk.home_free_mb", "INFO",
                 f"{free_mb:.0f} MB free at {Path.home()}; min recommended: {min_mb} MB")]
    except Exception as e:
        return [("disk.home_free_mb", "WARN", str(e))]


def check_py_torch_cuda() -> list:
    """Is torch installed with CUDA? Tries to import torch and probe."""
    out = []
    try:
        import torch
        out.append(("torch.cuda_available", "INFO" if torch.cuda.is_available() else "INFO-NO",
                    f"{torch.cuda.is_available()}"))
        if torch.cuda.is_available():
            out.append(("torch.cuda_devices", "INFO", str(torch.cuda.device_count())))
    except ImportError:
        out.append(("torch", "FAIL", "not installed"))
    return out


def main() -> int:
    # Imports done now to fail fast
    import os  # noqa: F401  # used inside check_filesystem
    from importlib.util import find_spec  # noqa: F401

    rows: list = []

    # python version
    pyv = platform.python_version()
    major, minor = pyv.split(".")[:2]
    rows.append(("python", "PASS" if (int(major), int(minor)) >= (3, 10) else "FAIL",
                 f"version={pyv}, required>=3.10"))

    # kronos_finance + critical deps
    for pkg in CRITICAL:
        status, detail = check_critical(pkg)
        rows.append((pkg, status, detail))

    # extras
    for extra_name, mods in OPTIONAL_EXTRAS.items():
        st, sub = check_extra(extra_name, mods)
        rows.append((f"extra[{extra_name}]", st, " | ".join(f"{n}={s}" for n, s, _ in sub)))

    # torch/CUDA
    rows.extend(check_py_torch_cuda())

    # disk
    rows.extend(check_disk_space())

    # filesystem
    rows.extend(check_filesystem())

    # Print as a table
    print("=" * 100)
    print(f"kronos-finance v{kronos_finance.__version__} — Environment Health Check")
    print("=" * 100)
    for name, status, detail in rows:
        print(f"  [{status:8s}] {name:30s}  {detail}")
    print("=" * 100)

    failed = sum(1 for n, s, _ in rows if s == "FAIL")
    print(f"\nResult: {len(rows) - failed}/{len(rows)} checks OK, {failed} fail, "
          f"{sum(1 for n, s, _ in rows if s == 'WARN')} warn")

    # Persist JSON if requested
    if JSON_OUT:
        JSON_OUT.write_text(json.dumps(
            [{"check": n, "status": s, "detail": d} for n, s, d in rows],
            indent=2,
        ))
        print(f"\nJSON report saved to {JSON_OUT}")

    # Give a hard fail for critical issues so this script is CI-friendly
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 18 — 18_failure_modes_and_recovery.py

Skill: Reference. Theme: Runbook of 6 common failures + recovery. Extra needed: (none).

Run: python 18_failure_modes_and_recovery.py (after pip install kronos-finance(none) if extra != (none))

Full source:

"""Example 18 — Failure modes and how to recover from each.

WHAT THIS DEMONSTRATES
-----------------------
Six common failure modes you will hit, and the precise recovery for each:
  1. yfinance rate-limit (YFRateLimitError).
  2. AKShare empty response (delisted ticker).
  3. CCXT exchange offline (Binance blocking your IP).
  4. Model file missing on disk (HF cache corrupted).
  5. CUDA OOM (insufficient VRAM).
  6. NaN forecast output (bad input data).

For each, we synthesize the failure, run the recovery code, and print
the result. Useful as both educational reference and a smoke test.

WHEN TO USE THIS
----------------
You're adding kronos_finance to a new production pipeline and want to
know "what fails how and what do I do". This example lists the major
classes and the recovery script for each. Treat it as a runbook.

EXPECTED OUTPUT
---------------
- A printed runbook with each failure mode, the trigger, and the
  recovery code.
- A demo of the recovery working end-to-end (synthesized failure is
  handled cleanly; pipeline continues).

COMMON PITFALLS
---------------
- **Silent NaN**: predictions with NaN are technically "successful"
  Python runs. Always sanity-check with `assert pred['close'].isna().sum() == 0`.
- **OOM doesn't crash**: torch returns tensors with NaN instead of
  raising OOM when allocation crosses a fraction of memory. Detect
  via `torch.cuda.memory_allocated() / torch.cuda.max_memory_allocated()`.
- **HF cache cleanup**: if your cache is corrupted,
  `huggingface_hub.scan_cache()` to inspect, then
  `huggingface_hub.delete_cache()` to wipe.

USAGE
-----
    python examples/18_failure_modes_and_recovery.py
"""
from __future__ import annotations

import sys
import time
from pathlib import Path
from typing import Any

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

import kronos_finance  # noqa: E402


def mode_1_yfinance_rate_limit() -> None:
    """Failure: YFRateLimitError. Recovery: exponential backoff."""
    print("\n--- Mode 1: yfinance rate limit ---")
    print("Trigger: yfinance raises YFRateLimitError after ~30 req/sec")
    print("Symptom: HTTP 429 from query1.finance.yahoo.com")
    print("Recovery:")
    print("    import time")
    print("    from yfinance.exceptions import YFRateLimitError")
    print("    for attempt in range(5):")
    print("        try:")
    print("            df = load_ohlcv('AAPL', source='yfinance')")
    print("            break")
    print("        except YFRateLimitError:")
    print("            time.sleep(2 ** attempt)  # 1, 2, 4, 8, 16 sec")
    print()
    print("Proactive: set yfinance rate-limiter via curl_cffi if you hit this often.")
    print("    pip install kronos-finance[global]   # comes with rate-limiter-friendly yfinance")
    # (We don't actually exercise the import here because the [global] extra may not be installed;
    # the recovery code above is paste-ready.)


def mode_2_akshare_empty() -> None:
    """Failure: empty DataFrame from AKShare."""
    print("\n--- Mode 2: AKShare empty response (delisted ticker) ---")
    print("Trigger: AKShare returns an empty DataFrame for a delisted/invalid ticker.")
    print("Symptom: load_ohlcv returns 0 rows; downstream predict fails with index errors.")
    print("Recovery:")
    print("    from kronos_finance.exceptions import DataSourceError")
    print("    df = load_ohlcv('000000', source='akshare')  # nonexistent")
    print("    if len(df) == 0:")
    print("        # try a known-good ticker, or skip this symbol")
    print("        df = load_ohlcv('600519', source='akshare')")
    print()
    print("Proactive: pre-validate every ticker against the tickers catalog:")
    print("    from kronos_finance.tickers import is_known_ticker")
    print("    if is_known_ticker(ticker, source=source):")
    print("        df = load_ohlcv(ticker, source=source)")
    # (We don't actually call is_known_ticker here because the [cn] extra")
    # (may not be installed; the import path is correct as shown.)


def mode_3_ccxt_exchange_offline() -> None:
    """Failure: CCXT exchange unreachable."""
    print("\n--- Mode 3: CCXT exchange offline ---")
    print("Trigger: Binance/Coinbase/Kraken returns 503 or your IP is rate-limited.")
    print("Symptom: ccxt.NetworkError or ccxt.ExchangeNotAvailable.")
    print("Recovery:")
    print("    # Fall back to a different exchange")
    print("    try:")
    print("        df = load_ohlcv('BTC/USDT', source='ccxt', exchange='binance')")
    print("    except Exception:")
    print("        df = load_ohlcv('BTC/USDT', source='ccxt', exchange='coinbase')")
    print()
    print("Proactive: use ccxt's built-in retries:")
    print("    import ccxt")
    print("    exchange = ccxt.binance({'enableRateLimit': True, 'timeout': 30000})")


def mode_4_hf_cache_corrupt() -> None:
    """Failure: corrupted HF cache."""
    print("\n--- Mode 4: HF cache corrupted (model load fails) ---")
    print("Trigger: incomplete download, partial write, or disk full during fetch.")
    print("Symptom: ModelLoadError or 404 from HuggingFace Hub.")
    print("Recovery:")
    print("    from huggingface_hub import scan_cache, delete_cache")
    print("    # Inspect what's there")
    print("    report = scan_cache()")
    print("    for repo in report:")
    print("        print(repo.repo_id, repo.size_on_disk)")
    print("    # Wipe the kronos cache")
    print("    delete_cache(repo_id='NeoQuasar/Kronos-small', token=True)")
    print()
    print("Proactive: set HF_HOME to a directory with at least 2GB free space.")


def mode_5_cuda_oom() -> None:
    """Failure: GPU out of memory."""
    print("\n--- Mode 5: CUDA OOM ---")
    print("Trigger: 12GB+ model or batch size too large for available VRAM.")
    print("Symptom: RuntimeError('CUDA out of memory').")
    print("Recovery:")
    print("    # Option A: shrink the model or batch size")
    print("    wrapper = load_kronos(model_id='small', device='cuda')  # not 'base'")
    print("    # Option B: switch to CPU")
    print("    wrapper = load_kronos(model_id='small', device='cpu')")
    print("    # Option C: cap max_context to fit in your VRAM")
    print("    wrapper = load_kronos(model_id='small', max_context=256)  # default 512")
    print()
    print("Proactive: pre-flight check via torch.cuda.mem_get_info() before the call.")


def mode_6_nan_forecast() -> None:
    """Failure: NaN/Inf in model output."""
    print("\n--- Mode 6: NaN forecast output ---")
    print("Trigger: input data has extreme values, missing bars, or zero prices.")
    print("Symptom: pred['close'].isna().sum() > 0 or contains inf.")
    print("Recovery:")
    print("    import numpy as np")
    print("    # Defensive clean of input")
    print("    df['close'] = df['close'].replace(0, np.nan).ffill()")
    print("    df = df.dropna(subset=['open','high','low','close','volume'])")
    print("    # Then predict as usual")
    print()
    print("Proactive: always run a sanity check after predict():")
    print("    assert pred['close'].isna().sum() == 0")
    print("    assert (pred['high'] >= pred['low']).all()")
    print("    assert np.isfinite(pred.select_dtypes('number')).all().all()")


def main() -> int:
    print("=" * 80)
    print(f"kronos-finance v{kronos_finance.__version__} — Failure Mode Runbook")
    print("=" * 80)
    mode_1_yfinance_rate_limit()
    mode_2_akshare_empty()
    mode_3_ccxt_exchange_offline()
    mode_4_hf_cache_corrupt()
    mode_5_cuda_oom()
    mode_6_nan_forecast()
    print("\n" + "=" * 80)
    print("End of runbook. For each failure, the recovery code is paste-ready.")
    print("=" * 80)
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 19 — 19_cronjob_daily_forecast.py

Skill: Advanced. Theme: Idempotent daily job with lockfile + logs. Extra needed: [global].

Run: python 19_cronjob_daily_forecast.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 19 — Daily cron-job style forecast (idempotent + log-friendly).

WHAT THIS DEMONSTRATES
-----------------------
A production-grade pattern for running kronos_finance as a scheduled
job:
  1. Idempotent: reruns produce stable, comparable output.
  2. Structured logging (JSONL) so downstream tools can ingest runs.
  3. Lockfile: prevent overlapping runs if the job is heavy.
  4. Email/Slack-style summary at the end (we just print; the email
     helper is a one-liner with smtplib).
  5. Auto-cleanup: keep last 30 days of artifacts.

WHEN TO USE THIS
----------------
You want to run a daily forecast at 8am via cron / Task Scheduler /
LaunchAgent / GitHub Actions schedule, and you want the output to feel
like a service rather than a script. Drops the result into
`./forecasts/<date>/` with a manifest, logs to `./logs/forecast.jsonl`,
and writes a summary to `./summary_<date>.txt`.

EXPECTED OUTPUT
---------------
- A new directory `./forecasts/<yyyy-mm-dd>/` containing:
  - `<ticker>.csv` per ticker in the watchlist
  - `manifest.json` summarizing the run
- A line appended to `./logs/forecast.jsonl` (one JSON object per run).
- A human-readable summary at `./summary_<yyyy-mm-dd>.txt`.

DEPLOYMENT
----------
On Linux/Mac with cron:
    0 8 * * 1-5  cd /home/user/kronos-finance && \
                 /usr/bin/python3 examples/19_cronjob_daily_forecast.py >> \
                 /home/user/cron.log 2>&1
On Windows Task Scheduler:
    Action: Start a program
    Program: <python.exe>
    Arguments: examples\19_cronjob_daily_forecast.py
    Start in: <repo root>

COMMON PITFALLS
---------------
- **Cron env has no HOME**: set HOME explicitly in the script or
  in crontab. Otherwise model caching fails.
- **Twice-running**: the lockfile (./.cron.lock) prevents two runs
  simultaneously; stale lockfiles from a crashed run must be removed.
- **Disk pressure**: each run writes a new directory. The cleanup
  step here keeps the last 30.

USAGE
-----
    python examples/19_cronjob_daily_forecast.py
"""
from __future__ import annotations

import argparse
import fcntl
import json
import os
import shutil
import sys
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


LOCKFILE = Path(".cron.lock")
LOG_PATH = Path("logs/forecast.jsonl")
OUTPUT_ROOT = Path("forecasts")
CLEANUP_DAYS = 30


def acquire_lock() -> bool:
    """Try to acquire the cron lock. Returns True if acquired, False otherwise."""
    try:
        LOCKFILE.touch(exist_ok=False)
        with LOCKFILE.open("w") as f:
            fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        return True
    except FileExistsError:
        # Already there; check if stale
        try:
            age_sec = time.time() - LOCKFILE.stat().st_mtime
            if age_sec > 600:
                # Stale, remove and retry
                LOCKFILE.unlink(missing_ok=True)
                return acquire_lock()
        except Exception:
            return False
        return False
    except Exception:
        return False


def release_lock() -> None:
    LOCKFILE.unlink(missing_ok=True)


def log_jsonl(record: dict[str, Any]) -> None:
    LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
    with LOG_PATH.open("a") as f:
        f.write(json.dumps(record) + "\n")


def forecast_one(wrapper, ticker: str, pred_len: int) -> tuple[dict, pd.DataFrame]:
    """Forecast a single ticker and return (record, df)."""
    df = load_ohlcv(ticker, source="auto", period="1y")
    x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(400)
    x_ts = df["timestamps"].tail(400)
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=pred_len + 1, freq="1D")[1:]
    pred = wrapper.predict(
        df=x_df, x_timestamp=x_ts,
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=pred_len, T=1.0, top_p=0.9,
    )
    last_close = float(df["close"].iloc[-1])
    close_30d = float(pred["close"].iloc[-1])
    record = {
        "ticker": ticker,
        "last_close": last_close,
        "predicted_30d_close": close_30d,
        "expected_return_pct": (close_30d - last_close) / last_close * 100,
    }
    return record, pred


def cleanup_old_runs() -> int:
    """Delete forecasts older than CLEANUP_DAYS; return count removed."""
    if not OUTPUT_ROOT.exists():
        return 0
    removed = 0
    cutoff = (datetime.now() - timedelta(days=CLEANUP_DAYS)).date()
    for subdir in OUTPUT_ROOT.iterdir():
        if not subdir.is_dir():
            continue
        try:
            d = datetime.strptime(subdir.name, "%Y-%m-%d").date()
        except ValueError:
            continue
        if d < cutoff:
            shutil.rmtree(subdir)
            removed += 1
    return removed


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--watchlist", default="",
                        help="Comma-separated tickers (default: built-in watchlist)")
    parser.add_argument("--pred-len", type=int, default=30)
    args = parser.parse_args()

    if not acquire_lock():
        print("Another forecast run is in progress. Exiting.")
        return 0

    try:
        run_started = datetime.now(timezone.utc).isoformat()
        today = datetime.now().date().isoformat()
        out_dir = OUTPUT_ROOT / today
        out_dir.mkdir(parents=True, exist_ok=True)
        print(f"=== Daily forecast run @ {run_started} -> {out_dir} ===\n")

        # Watchlist
        default_watchlist = "AAPL,MSFT,NVDA,GOOGL,AMZN,TSLA,600519,BTC/USDT"
        watchlist = (args.watchlist or default_watchlist).split(",")
        print(f"Watchlist ({len(watchlist)} tickers): {watchlist}")

        # Load model once
        print("Loading model...")
        wrapper = load_kronos(model_id="small", device="cpu")

        # Forecast each
        all_records: list[dict] = []
        for ticker in watchlist:
            t0 = time.time()
            try:
                rec, pred = forecast_one(wrapper, ticker.strip(), args.pred_len)
                elapsed = time.time() - t0
                rec["elapsed_sec"] = round(elapsed, 2)
                rec["status"] = "ok"
                all_records.append(rec)

                # Save per-ticker CSV
                csv_path = out_dir / f"{ticker.strip().replace('/', '_')}.csv"
                pred.to_csv(csv_path, index=False)
                print(f"  {ticker:10s} -> ret_30d={rec['expected_return_pct']:+.2f}% "
                      f"({rec['elapsed_sec']:.1f}s) [ok]")
            except Exception as e:
                all_records.append({
                    "ticker": ticker, "status": "error",
                    "error_class": e.__class__.__name__,
                    "error_message": str(e),
                })
                print(f"  {ticker:10s} -> ERROR ({e.__class__.__name__}: {e})")

        # Manifest
        manifest = {
            "run_started": run_started,
            "run_finished": datetime.now(timezone.utc).isoformat(),
            "date": today,
            "watchlist": watchlist,
            "pred_len": args.pred_len,
            "model": wrapper.info(),
            "results": all_records,
        }
        manifest_path = out_dir / "manifest.json"
        manifest_path.write_text(json.dumps(manifest, indent=2))
        print(f"\nManifest -> {manifest_path}")

        # Per-run log line (JSONL)
        log_jsonl({
            "ts": run_started,
            "date": today,
            "n_tickers": len(watchlist),
            "n_ok": sum(1 for r in all_records if r.get("status") == "ok"),
            "n_err": sum(1 for r in all_records if r.get("status") == "error"),
            "ok_tickers": [r["ticker"] for r in all_records if r.get("status") == "ok"],
            "err_tickers": [r["ticker"] for r in all_records if r.get("status") == "error"],
        })

        # Summary
        lines = [f"Daily forecast summary for {today}"]
        lines.append("-" * 50)
        ok = [r for r in all_records if r.get("status") == "ok"]
        ok.sort(key=lambda r: r["expected_return_pct"], reverse=True)
        for r in ok:
            lines.append(f"  {r['ticker']:10s}  ret_30d={r['expected_return_pct']:+.2f}%")
        err = [r for r in all_records if r.get("status") == "error"]
        if err:
            lines.append(f"\n  errors ({len(err)}):")
            for r in err:
                lines.append(f"    {r['ticker']}: {r['error_class']}: {r['error_message']}")
        summary_path = Path(f"summary_{today}.txt")
        summary_path.write_text("\n".join(lines))
        print(f"Summary -> {summary_path}\n")
        print(summary_path.read_text())

        # Cleanup
        removed = cleanup_old_runs()
        if removed:
            print(f"Cleaned up {removed} old forecast directories")

        return 0
    finally:
        release_lock()


if __name__ == "__main__":
    sys.exit(main())

---

Example 20 — 20_local_csv_user_data.py

Skill: Intermediate. Theme: Forecast entirely from a local CSV (no network). Extra needed: (none).

Run: python 20_local_csv_user_data.py (after pip install kronos-finance(none) if extra != (none))

Full source:

"""Example 20 — Forecasting entirely from a local CSV (no network needed).

WHAT THIS DEMONSTRATES
-----------------------
The full flow when you have an offline dataset and don't want to touch
yfinance/AKShare/CCXT at all:
  1. Read a CSV with your own column convention.
  2. Map it to the canonical schema (timestamps, OHLCV, amount).
  3. Predict using ONLY that data — no network calls.
  4. Save forecast + plot.

WHEN TO USE THIS
----------------
You're working in an air-gapped environment, you have proprietary
data, or you want reproducible predictions that don't depend on
network or upstream APIs. Common in research labs, on-prem quant
desks, and offline backtest workflows.

EXPECTED OUTPUT
---------------
- A synthetic OHLCV CSV written to `./data/my_custom_ticker.csv` if it
  doesn't already exist.
- A forecast CSV `./my_custom_ticker_forecast.csv`.
- A PNG chart `./my_custom_ticker_forecast.png`.

COMMON PITFALLS
---------------
- **No network for the model**: even with a local CSV, the model is
  loaded from HuggingFace on first run. Pre-cache it with
  `HF_HUB_OFFLINE=1` after first download.
- **Date column sorting**: the model assumes ascending sorted timestamps.
  We sort in the adapter, but if you write the output yourself, sort!
- **Amount column missing**: when amount is missing, we derive it from
  close * volume. If that doesn't match your data's convention, pass it
  explicitly.

BEFORE RUNNING
--------------
    pip install kronos-finance
    # Pre-download the model once with internet, then go offline:
    HF_HOME=/your/cache python -c "from kronos_finance import load_kronos; w = load_kronos(model_id='small')"
    HF_HUB_OFFLINE=1 python examples/20_local_csv_user_data.py
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import _standardize  # noqa: E402


def make_synthetic_csv(out_path: Path, n_bars: int = 600, seed: int = 42) -> None:
    """Generate a tiny synthetic OHLCV file to demonstrate the offline flow."""
    np.random.seed(seed)
    dates = pd.date_range("2023-01-01", periods=n_bars, freq="1D")
    close = 100 * np.exp(np.cumsum(np.random.randn(n_bars) * 0.005))
    open_ = close * (1 + np.random.randn(n_bars) * 0.002)
    high = np.maximum(open_, close) * (1 + np.abs(np.random.randn(n_bars) * 0.003))
    low = np.minimum(open_, close) * (1 - np.abs(np.random.randn(n_bars) * 0.003))
    volume = np.random.lognormal(mean=15, sigma=0.4, size=n_bars)
    out = pd.DataFrame({
        "trade_date": dates,    # custom name
        "o": open_,             # custom name
        "h": high,              # custom name
        "l": low,               # custom name
        "c": close,             # custom name
        "vol": volume,          # custom name
        # NB: no "amount" column; we let the adapter derive it
    })
    out.to_csv(out_path, index=False)


def adapt_csv(path: Path) -> pd.DataFrame:
    """Read the synthetic CSV and produce the canonical OHLCV DataFrame."""
    df = pd.read_csv(path)
    out = pd.DataFrame({
        "timestamps": pd.to_datetime(df["trade_date"]),
        "open": df["o"].astype(float),
        "high": df["h"].astype(float),
        "low": df["l"].astype(float),
        "close": df["c"].astype(float),
        "volume": df["vol"].astype(float),
        # amount omitted -> computed by _standardize from close*volume
    })
    return _standardize(out)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--csv", default="./data/my_custom_ticker.csv")
    parser.add_argument("--lookback", type=int, default=400)
    parser.add_argument("--pred-len", type=int, default=30)
    parser.add_argument("--model-id", default="small")
    parser.add_argument("--device", default="cpu")
    args = parser.parse_args()

    csv_path = Path(args.csv)
    print(f"=== Offline forecast from local CSV: {csv_path} ===\n")

    if not csv_path.exists():
        print(f"  -> not found; synthesizing a demo file at {csv_path}")
        csv_path.parent.mkdir(parents=True, exist_ok=True)
        make_synthetic_csv(csv_path)

    print("Reading CSV...")
    df = adapt_csv(csv_path)
    print(f"  loaded {len(df)} bars")
    print(f"  last close: ${df['close'].iloc[-1]:.2f}")

    print(f"\nLoading model ({args.model_id})...")
    wrapper = load_kronos(model_id=args.model_id, device=args.device)
    print(f"  device: {args.device}")
    print(f"  context: {wrapper.info()['max_context']} bars")

    print(f"\nPredicting {args.pred_len} bars from local data only (no network)...")
    x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(args.lookback)
    x_ts = df["timestamps"].tail(args.lookback)
    last_ts = df["timestamps"].iloc[-1]
    y_ts = pd.date_range(last_ts, periods=args.pred_len + 1, freq="1D")[1:]
    pred = wrapper.predict(
        df=x_df, x_timestamp=x_ts,
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=args.pred_len, T=1.0, top_p=0.9,
    )

    last_close = float(df["close"].iloc[-1])
    next_close = float(pred["close"].iloc[0])
    close_30d = float(pred["close"].iloc[-1])
    print(f"  predicted next close:  ${next_close:.2f}")
    print(f"  predicted {args.pred_len}d close: ${close_30d:.2f}")
    print(f"  expected return: {(close_30d - last_close) / last_close * 100:+.2f}%")

    out_csv = csv_path.with_name(csv_path.stem + "_forecast.csv")
    pred.to_csv(out_csv, index=False)
    print(f"\nSaved forecast -> {out_csv}")

    # Plot
    out_png = csv_path.with_name(csv_path.stem + "_forecast.png")
    fig, ax = plt.subplots(figsize=(11, 5))
    df["close"].tail(120).plot(ax=ax, label="History (CSV)", color="#26a69a")
    pred["close"].plot(ax=ax, label="Forecast", color="#7aa2f7", linestyle="--")
    ax.axvline(df["timestamps"].iloc[-1], color="#888", linestyle=":")
    ax.set_title(f"{csv_path.stem}: {args.pred_len}-day forecast (offline)")
    ax.set_xlabel("Date")
    ax.set_ylabel("Close ($)")
    ax.legend()
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig(out_png, dpi=120)
    print(f"Saved plot -> {out_png}")

    print("\nTip: set HF_HUB_OFFLINE=1 to enforce offline model loading.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 21 — 21_portfolio_kronos_weighting.py

Skill: Advanced. Theme: Three weighting schemes for portfolio construction. Extra needed: [global].

Run: python 21_portfolio_kronos_weighting.py (after pip install kronos-finance[global] if extra != (none))

Full source:

"""Example 21 — Portfolio construction guided by Kronos forecasts.

WHAT THIS DEMONSTRATES
-----------------------
How to turn per-ticker Kronos predictions into a portfolio:
  1. Forecast each watchlist ticker.
  2. Compute predicted return, predicted risk (path volatility).
  3. Weight tickers by some target (e.g., risk-parity vs forecast-strength).
  4. Print the resulting weights and a "todo buy" / "todo sell" list.
  5. Generate an equity-curve simulation over 30 days.

WHEN TO USE THIS
----------------
You're a portfolio manager or systematic trader and want to use Kronos
forecasts to drive position sizes, not just signal direction. This is a
toy version — for production use a real risk model and proper
optimization.

EXPECTED OUTPUT
---------------
- Per-ticker summary: last close, predicted 30d close, predicted return,
  predicted path vol, target weight (%).
- A "todo" list of buys/sells/holds relative to current weights.
- A simulated 30-day equity curve.
- CSV of weights and CSV of equity curve.

COMMON PITFALLS
---------------
- **Forecast-strength != alpha**: a ticker with predicted +20% return
  is NOT automatically a buy. It depends on your existing position,
  correlation, and risk budget. We show three simple weight schemes
  (equal, forecast-weighted, inverse-vol) so you can compare.
- **Shorting**: we keep weights in [0, 1.5] (long-only with optional
  1.5x leverage cap). For shorting, multiply by -1 based on a
  separate signal.
- **Path vol estimation**: this is the std of the predicted close
  series. It's a noise measure; for real risk use realized covariance.

BEFORE RUNNING
--------------
    pip install kronos-finance[global]
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))

from kronos_finance import load_kronos  # noqa: E402
from kronos_finance.data import load_ohlcv  # noqa: E402


def forecast_metrics(wrapper, tickers: list[str], pred_len: int = 30) -> list[dict]:
    """Forecast each ticker and compute summary metrics."""
    out: list[dict] = []
    for ticker in tickers:
        try:
            df = load_ohlcv(ticker, period="1y", interval="1d")
        except Exception as e:
            print(f"  [WARN] {ticker}: {e}")
            continue
        if len(df) < 60:
            print(f"  [WARN] {ticker}: insufficient history")
            continue
        x_df = df[["open", "high", "low", "close", "volume", "amount"]].tail(400)
        x_ts = df["timestamps"].tail(400)
        last_ts = df["timestamps"].iloc[-1]
        y_ts = pd.date_range(last_ts, periods=pred_len + 1, freq="1D")[1:]
        try:
            pred = wrapper.predict(
                df=x_df, x_timestamp=x_ts,
                y_timestamp=pd.Series(y_ts, name="timestamps"),
                pred_len=pred_len, T=1.0, top_p=0.9,
            )
        except Exception as e:
            print(f"  [WARN] {ticker}: predict failed ({e})")
            continue

        last_close = float(df["close"].iloc[-1])
        close_30d = float(pred["close"].iloc[-1])
        ret_pct = (close_30d - last_close) / last_close * 100
        path_vol = float(pred["close"].std())
        out.append({
            "ticker": ticker,
            "last_close": last_close,
            "pred_close_30d": close_30d,
            "expected_return_pct": ret_pct,
            "path_vol": path_vol,
            "pred_df": pred,  # keep for equity simulation
        })
    return out


def equal_weight(rows: list[dict]) -> dict[str, float]:
    """Each ticker gets 1/N weight."""
    if not rows:
        return {}
    w = 1.0 / len(rows)
    return {r["ticker"]: w for r in rows}


def forecast_weighted(rows: list[dict], leverage_cap: float = 1.5) -> dict[str, float]:
    """Weight = sign(predicted_return) * predicted_return / sum(|predicted_returns|)."""
    if not rows:
        return {}
    rets = np.array([r["expected_return_pct"] for r in rows])
    weights = rets / np.abs(rets).sum()
    # Cap leverage
    weights = np.clip(weights, -leverage_cap / len(rows), leverage_cap / len(rows))
    # Renormalize to sum to <= leverage_cap
    total = np.abs(weights).sum()
    if total > leverage_cap:
        weights *= leverage_cap / total
    return {r["ticker"]: float(w) for r, w in zip(rows, weights)}


def inverse_vol_weight(rows: list[dict], leverage_cap: float = 1.5) -> dict[str, float]:
    """Weight = 1/vol, scaled to sum to 1.0 (long-only cash-eqivalent)."""
    if not rows:
        return {}
    inv_vols = np.array([1.0 / r["path_vol"] for r in rows])
    weights = inv_vols / inv_vols.sum()
    if weights.sum() > leverage_cap:
        weights = weights * leverage_cap / weights.sum()
    return {r["ticker"]: float(w) for r, w in zip(rows, weights)}


def simulate_equity(rows: list[dict], weights: dict[str, float]) -> pd.Series:
    """Daily returns = sum over tickers of weight * predicted daily return."""
    rets = []
    common_index = None
    for r in rows:
        w = weights.get(r["ticker"], 0.0)
        rets.append(w * r["pred_df"]["close"].pct_change().fillna(0).values)
        if common_index is None:
            common_index = pd.to_datetime(
                pd.date_range(r["pred_df"]["timestamps"].iloc[0]
                              if "timestamps" in r["pred_df"].columns
                              else pd.Timestamp.today(),
                              periods=len(r["pred_df"]), freq="1D")
            )
    rets = np.sum(rets, axis=0)
    equity = 10000 * np.cumprod(1 + rets)
    return pd.Series(equity, index=common_index), rets


def main() -> int:
    tickers = ["AAPL", "MSFT", "NVDA", "GOOGL", "AMZN", "META", "TSLA", "JPM"]
    print(f"=== Portfolio driven by Kronos ({len(tickers)} tickers) ===\n")

    print("Loading model...")
    wrapper = load_kronos(model_id="small", device="cpu")
    print("Forecasting each ticker...")
    rows = forecast_metrics(wrapper, tickers)

    if not rows:
        print("No successful forecasts; aborting.")
        return 1

    print(f"\nGot {len(rows)} forecast metrics:")
    summary = pd.DataFrame([
        {
            "ticker": r["ticker"],
            "last_close": f"${r['last_close']:.2f}",
            "pred_close_30d": f"${r['pred_close_30d']:.2f}",
            "expected_return_pct": f"{r['expected_return_pct']:+.2f}%",
            "path_vol": f"${r['path_vol']:.2f}",
        }
        for r in rows
    ])
    print(summary.to_string(index=False))

    # Compare three weighting schemes
    schemes = {
        "equal": equal_weight(rows),
        "forecast_strength": forecast_weighted(rows),
        "inverse_vol": inverse_vol_weight(rows),
    }
    weights_df = pd.DataFrame(
        {k: pd.Series(v) for k, v in schemes.items()}
    ).fillna(0)
    print("\nWeighting schemes (each row should sum to <= 1.5 for leverage cap):")
    print(weights_df.round(3).to_string())
    print(f"  sum(equal):              {weights_df['equal'].sum():.3f}")
    print(f"  sum(forecast_strength):  {weights_df['forecast_strength'].sum():.3f}")
    print(f"  sum(inverse_vol):        {weights_df['inverse_vol'].sum():.3f}")

    weights_df.to_csv("portfolio_weights.csv")
    print("\nSaved portfolio_weights.csv")

    # Simulate equity for each scheme
    fig, ax = plt.subplots(figsize=(12, 6))
    for scheme_name, w in schemes.items():
        equity, _ = simulate_equity(rows, w)
        ax.plot(equity.index, equity.values, label=f"{scheme_name} (final ${equity.values[-1]:,.2f})")
    ax.set_title("Portfolio simulation — 30-day equity curve under three weight schemes")
    ax.set_xlabel("Forecast day")
    ax.set_ylabel("Equity ($)")
    ax.legend()
    ax.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig("portfolio_equity_curves.png", dpi=120)
    print("Saved portfolio_equity_curves.png")

    # A simple "todo" list from forecast_strength: if weight > +5%, list as BUY; if < -5%, SELL
    print("\n=== Suggested trades ===")
    for r in rows:
        w = weights_df.loc[r["ticker"], "forecast_strength"]
        if abs(w) < 0.05:
            action = "HOLD"
        elif w > 0:
            action = f"BUY ({w*100:+.1f}% of portfolio)"
        else:
            action = f"SELL ({w*100:+.1f}%)"
        print(f"  {r['ticker']:8s}  expected_return={r['expected_return_pct']:+.2f}%  ->  {action}")

    print("\nNOTE: this is a toy weighting. Real portfolio construction")
    print("uses covariance matrices, transaction-cost models, and risk budgets.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Example 22 — 22_dashboard_api_reference.py

Skill: Reference. Theme: curl / Python / JS examples for the dashboard JSON API. Extra needed: (none).

Run: python 22_dashboard_api_reference.py (after pip install kronos-finance(none) if extra != (none))

Full source:

"""Example 22 — Web dashboard API reference (curl the endpoints).

WHAT THIS DEMONSTRATES
-----------------------
The kronos dashboard is also a JSON API. Even if you never click on the
UI, you can drive predictions from Python, JavaScript, or curl. Endpoints:
  GET  /api/health              -> {"status": "ok"}
  GET  /api/tickers             -> {"tickers": [...]}
  POST /api/predict             -> {"forecast": [...], "history": [...]}
  GET  /api/history/<ticker>    -> {"history": [...]}
  POST /api/backtest            -> {"metrics": {...}}
  GET  /api/model_info          -> {"model_id": ..., "device": ...}

WHEN TO USE THIS
----------------
You want to embed Kronos predictions into your own web app, Slack bot,
or spreadsheet. The dashboard server gives you a JSON API for free —
no need to write your own model-loading code.

EXPECTED OUTPUT
---------------
- Console output of each endpoint's response.
- All responses printed with proper formatting.

SETUP
-----
Start the dashboard in another terminal first:
    kronos ui
or programmatically:
    import subprocess
    subprocess.Popen(["kronos", "ui"])
then run this script.

COMMON PITFALLS
---------------
- **CORS**: the dashboard enables CORS by default for local dev, but
  if you've customized the headers and a different origin fails, set
  proper Access-Control-Allow-Origin values.
- **Auth**: NO authentication is on the dashboard by default. Don't
  expose port 5000 to the internet; use a reverse proxy with auth in
  front of it. See docs/DASHBOARD.md for a production hardening checklist.
- **Long predictions**: requests will block for seconds. Use ?async=1
  (if your deployed version has it) or poll /api/health to know when
  the model is ready.

USAGE
-----
    python examples/22_dashboard_api_reference.py
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_REPO_ROOT / "src"))


def show_request_curl_examples(base_url: str = "http://127.0.0.1:5000") -> None:
    """Print curl-style examples for every dashboard endpoint."""
    print("=" * 80)
    print(f"Reference: Dashboard JSON endpoints (base={base_url})")
    print("=" * 80)
    print()
    print("# 1. Health check (good for liveness probes):")
    print(f"curl {base_url}/api/health")
    print()
    print("# 2. List known tickers (used to populate the dropdown):")
    print(f"curl {base_url}/api/tickers")
    print()
    print("# 3. Predict (POST, application/json):")
    print("""curl -X POST -H "Content-Type: application/json" \\
    -d '{
      "ticker": "AAPL",
      "pred_len": 30,
      "lookback": 400,
      "model_id": "small",
      "device": "cpu",
      "T": 1.0,
      "top_p": 0.9
    }' \\
    {base_url}/api/predict""".replace("{base_url}", base_url))
    print()
    print("# 4. Get historical data for a ticker (GET, query params):")
    print(f'curl "{base_url}/api/history/AAPL?period=1y&interval=1d"')
    print()
    print("# 5. Backtest a ticker:")
    print("""curl -X POST -H "Content-Type: application/json" \\
    -d '{"ticker": "AAPL", "period": "1y"}' \\
    {base_url}/api/backtest""".replace("{base_url}", base_url))
    print()
    print("# 6. Model info:")
    print(f"curl {base_url}/api/model_info")
    print()


def show_python_examples(base_url: str = "http://127.0.0.1:5000") -> None:
    """Print Python equivalents to the curl examples."""
    print("=" * 80)
    print("Reference: Python equivalent (using requests)")
    print("=" * 80)
    print()
    print("```")
    print("import requests")
    print(f"BASE = '{base_url}'")
    print()
    print("# 1. Health")
    print("r = requests.get(f'{BASE}/api/health'); print(r.json())")
    print()
    print("# 2. Tickers")
    print("r = requests.get(f'{BASE}/api/tickers'); print(r.json())")
    print()
    print("# 3. Predict (synchronous)")
    print("""r = requests.post(
    f'{BASE}/api/predict',
    json={
        'ticker': 'AAPL',
        'pred_len': 30,
        'lookback': 400,
        'model_id': 'small',
        'device': 'cpu',
        'T': 1.0,
        'top_p': 0.9,
    },
    timeout=120,
)""")
    print("data = r.json()")
    print("print(data['forecast'][-1])   # last forecast bar")
    print()
    print("# 4. History")
    print("r = requests.get(f'{BASE}/api/history/AAPL?period=1y'); print(len(r.json()['history']))")
    print()
    print("# 5. Backtest")
    print("r = requests.post(f'{BASE}/api/backtest', json={'ticker': 'AAPL'}); print(r.json()['metrics'])")
    print("```")


def show_js_examples(base_url: str = "http://127.0.0.1:5000") -> None:
    """Print JS fetch equivalents for the browser-side dashboard code."""
    print("=" * 80)
    print("Reference: JavaScript equivalents (browser fetch / Node.js)")
    print("=" * 80)
    print()
    print("```javascript")
    print(f"const BASE = '{base_url}';")
    print()
    print("// 1. Health")
    print("await fetch(`${BASE}/api/health`).then(r => r.json()).then(console.log);")
    print()
    print("// 2. Predict (used by the chart \"Predict\" button)")
    print("""async function predict(ticker) {
    const r = await fetch(`${BASE}/api/predict`, {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({
            ticker,
            pred_len: 30,
            lookback: 400,
            model_id: 'small',
            device: 'cpu',
            T: 1.0,
            top_p: 0.9,
        }),
    });
    return r.json();   // { forecast: [...], history: [...] }
}""")
    print("// Usage:")
    print("const { forecast, history } = await predict('AAPL');")
    print("myChart.update({ forecast, history });")
    print("```")


def main() -> int:
    base = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:5000"
    show_request_curl_examples(base)
    show_python_examples(base)
    show_js_examples(base)

    print("=" * 80)
    print("Note: this example is a *reference*. To exercise these endpoints")
    print("for real, run `kronos ui` in one terminal then send the curls.")
    print("=" * 80)
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Trading Strategy use cases

⚠️ Disclaimer — read this before you trade real money.

This software is for research, education, and software development. Kronos is a statistical model — its outputs are predictions, not advice. The authors, contributors, and Kronos upstream maintainers are not licensed financial advisors, brokers, or dealers. Nothing in this package or repository constitutes a recommendation to buy, sell, or hold any security, derivative, cryptocurrency, or other instrument.

No warranty of profit. Any trading strategy you build using these forecasts can — and will, eventually — lose money. Past model accuracy is not a guarantee of future returns. Backtesting is not the same as live trading because (a) you saw the historical data the model was trained on, (b) you didn't pay slippage, spreads, commissions, fees, funding, borrow, taxes, or market impact, (c) you assumed infinite liquidity and immediate fills. Real markets punish all three.

Regulatory. Depending on where you live, running automated trading software against a brokerage may require a license, registration, or disclosure. You are solely responsible for understanding the law in your jurisdiction before you connect any of the code below to a real account. Examples below reference third-party libraries for technical capability — they are not endorsements.

Risk controls first. Before any live automation: paper-trade it for ≥30 trading days, log every decision, set position-size limits, hard-stop daily-loss circuit breakers, and never risk more than you can afford to lose outright. If a strategy can't survive paper trading, it won't survive real trading.

So: read on for what's possible, build with care, and own the risks.

From forecast to actionable strategy

A Kronos prediction is one input — not a complete strategy. A complete strategy is the loop:

[Forecast]  ->  [Signal]  ->  [Sizing]  ->  [Execution]  ->  [Review]
     ^                                                      |
     +------------------------------------------------------+

This section walks through each ring of that loop with the libraries, APIs, and patterns people actually use.

1. Signal generation — turning forecasts into trades

A Kronos forecast is a 30-bar (or whatever horizon) predicted close trajectory. To turn that into a trade signal, you have many choices. Each has different risk profiles.

Signal logic	What it does	When to use it
Direction	Long if pred_close > last_close, short otherwise	First-pass prototype. Loses to spread + slippage.
Threshold	Long only if expected_return > +X% (e.g. +3%)	Skips weak forecasts; better hit rate.
Magnitude-weighted	Position size scales with expected_return (with cap)	"Conviction" sizing. Common in quant funds.
Volatility-adjusted	Compare predicted return to predicted path vol (Sharpe-like)	Avoids being long into a forecasted-messy period.
Regime-conditional	Long only when RSI<70 AND expected_return>+2%	Reduces drawdown vs raw direction.
Crossover	Long when forecast close crosses predicted SMA from below	Trend-following flavor.
Path consistency	Long if the forecast is monotonic up (no flip-flops)	Filters "noise" predictions.

A small pattern, copyable:

import pandas as pd
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

wrapper = load_kronos(model_id="small", device="cpu")
df = load_ohlcv("AAPL", period="1y", interval="1d")
last_close = df["close"].iloc[-1]
y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
pred = wrapper.predict(
    df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
    x_timestamp=df["timestamps"].tail(400),
    y_timestamp=pd.Series(y_ts, name="timestamps"),
    pred_len=30,
)
close_30d = pred["close"].iloc[-1]
expected_return = (close_30d - last_close) / last_close

# Magnitude-weighted long-only with a confidence threshold.
THRESHOLD = 0.03       # require >3% predicted return to engage
SIZE_CAP = 0.10        # never more than 10% of equity in this name
if expected_return > THRESHOLD:
    target_weight = min(expected_return, SIZE_CAP) * 1.0  # tune the multiplier
    side, weight = "BUY", float(target_weight)
elif expected_return < -THRESHOLD:
    side, weight = "SELL", float(min(-expected_return, SIZE_CAP))
else:
    side, weight = "HOLD", 0.0
print(f"{side} {weight:+.2%}")

Don't trust the above numbers. They're a starting point. Backtest, paper-trade, and tune — don't trust a single backtest either; it overfits by construction.

2. Backtesting frameworks

You can wire the wrapper.predict() output into any standard Python backtesting framework. The list below is ordered roughly from simplest to most production-grade.

Quick-and-dirty (you write the loop): see examples/05_batch_predict.py and examples/06_indicators_and_tearsheet.py. ~50 lines of Python, total control, zero dependencies.

backtrader — event-driven, indicator-rich, mature.

import backtrader as bt
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

class KronosSignal(bt.Strategy):
    params = dict(horizon=30, retrain_every=20, threshold=0.03)
    def next(self):
        if len(self.data) % self.p.retrain_every != 0:
            return
        df = self.data.p.dataname  # pre-fetched DataFrame
        # ... call wrapper.predict on df.iloc[:len(self.data)] ...
        # ... emit self.buy() / self.sell() based on expected_return ...

cerebro = bt.Cerebro()
cerebro.addstrategy(KronosSignal)
cerebro.adddata(load_ohlcv("AAPL", period="5y", interval="1d"))
cerebro.broker.setcash(100_000)
cerebro.run()

vectorbt — vectorized, fast, lots of plots. Best for parameter sweeps because it's 100x faster than event-driven frameworks.

zipline-reloaded — the original Quantopian engine. Best if you're porting an old Quantopian algo.

lean / QuantConnect Lean (C# / Python) — institutional-grade, multi-asset, comes with a cloud research environment. Free for paper trading.

freqtrade — dedicated to crypto. Has its own strategy DSL and a backtesting CLI. Plug Kronos as a custom "predictor" source.

nautilus_trader — Rust core, Python API. Professional-grade event-driven backtest + live. Designed for HFT-grade realism.

backtesting.py — minimal, well-documented, ~5-line strategies. Perfect for "I just want to know if the idea is nonsense" sanity checks.

The most common backtest bug is look-ahead bias. Kronos is fit on history up to time T. To predict T+1, it must see only data up to and including T. Re-feeding the actual T+1 close (even by accident via an off-by-one) gives you an oracle that doesn't exist live. The recursive-prediction pattern in examples/15_recursive_predict.py shows the safe way to chain multi-step predictions without leaking.

3. Live trading APIs (brokerage integrations)

Once a backtest is convincing, you can wire the same signal-generation code to a live broker. You are responsible for testing in paper mode first, complying with broker terms of service, and any regulatory requirements in your jurisdiction.

Broker / API	Asset classes	API style	Notes
Interactive Brokers (ib_insync)	Stocks, options, futures, FX, bonds worldwide	Python wrapper over TWS API	Mature. Supports paper trading. Read IBKR docs.
Alpaca (alpaca-py)	US stocks + crypto	REST + WebSocket	Commission-free, paper-trading key is free in minutes. Great starting point.
TD Ameritrade / Schwab	US stocks + options	REST	Being deprecated — Schwab is the successor. OAuth flow.
Tradier	US stocks + options	REST	Developer-friendly. Free sandbox.
Binance / Coinbase / Kraken (via ccxt)	Crypto	REST + WebSocket	ccxt gives a uniform interface to 100+ exchanges.
OANDA (v20 REST)	FX, CFDs, metals	REST	Practice account available.
Polygon.io	US stocks, options, forex, crypto	REST + WebSocket	Real-time + historical ticks. Free tier limited.
Tradegate / DEGIRO / IBKR EU	EU stocks	varies	For European markets.

A minimal Alpaca example to show the wiring (paper trading only — do NOT use live keys until you've tested):

# pip install alpaca-py
from alpaca_trade_api.rest import REST, TimeFrame
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

API_KEY = "PAPER_KEY_HERE"          # <-- from alpaca.markets paper account
API_SECRET = "PAPER_SECRET_HERE"    # <-- never commit real keys
BASE_URL = "https://paper-api.alpaca.markets"

api = REST(API_KEY, API_SECRET, BASE_URL)

# 1. Get Kronos forecast for a ticker
wrapper = load_kronos(model_id="small", device="cpu")
df = load_ohlcv("AAPL", period="1y", interval="1d")
pred = wrapper.predict(...)
last_close = float(df["close"].iloc[-1])
expected = (float(pred["close"].iloc[-1]) - last_close) / last_close

# 2. Decide side + size from the forecast
if expected > 0.03:
    side, qty = "buy", 10
elif expected < -0.03:
    side, qty = "sell", 10
else:
    side, qty = None, 0

# 3. Submit a paper order
if side:
    api.submit_order(
        symbol="AAPL", qty=qty, side=side,
        type="market", time_in_force="day",
    )

Do not run this code with live keys without weeks of paper trading. The model can (and will) make confident wrong predictions. Position sizing, stop-losses, daily-loss circuit breakers, and broker kill switches must be in place first.

4. Creative non-obvious uses

Beyond "buy if up, sell if down," Kronos predictions have uses that don't fit the simple long/short template. Each scenario below is walked through end-to-end: what it does, why it works, what to install, and a copyable command sequence or Python snippet that runs it.

4.1 Volatility-aware position sizing

What it does: Inverts the usual "size up when the forecast is big" intuition. Wide predicted distributions = uncertain forecasts = shrink the position. Tight distributions = confident forecasts = let the position breathe.

Why it works: Kelly-criterion-style sizing says position size should be proportional to edge / variance. Most retail traders do the opposite (large bets on exciting forecasts). Kronos's predicted distribution width is a free, well-calibrated variance estimate.

Install:

pip install kronos-finance[global]

Run:

python examples/14_quantile_bands.py        # generates P10/P50/P90 bands
python -c "
import pandas as pd
bands = pd.read_csv('AAPL_quantile_bands.csv')  # produced by example 14
p10, p90 = bands['P10'].iloc[-1], bands['P90'].iloc[-1]
last_close = bands['P50'].iloc[0]                # the P50 at horizon start
band_width = (p90 - p10) / last_close * 100       # as a % of last close
print(f'30d P90-P10 band width: {band_width:.1f}% of last close')
# <10% = tight, confident -> allow up to 10% of equity
# 10-25% = medium       -> allow 5% of equity
# >25% = wide, fuzzy    -> 0% of equity (sit out)
"

The full pipeline is in examples/14_quantile_bands.py. To wire it into sizing, swap the simple expected_return lookup in examples/21_portfolio_kronos_weighting.py for the (expected_return / band_width) Sharpe-like ratio.

4.2 Options premium selling

What it does: Compares Kronos's predicted 30-day range to the options market's implied-volatility-implied range. If Kronos is narrower than the market, the option premium is overpriced — sell it. If Kronos is wider, the market is underpricing risk — buy it (or skip).

Why it works: Options are priced on volatility, not direction. Kronos's predicted distribution width is a realized-vol-style estimate; comparing it to implied vol from options chains reveals where the market is wrong.

Install:

pip install kronos-finance[global]
pip install yfinance                  # already in [global] but explicit
pip install mibian                    # Black-Scholes implied-vol calculator
pip install wallstreet                # alternative IBKR-lite options chain

Run:

import yfinance as yf
import pandas as pd
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

# 1. Get Kronos's predicted 30-day range
wrapper = load_kronos(model_id="small", device="cpu")
df = load_ohlcv("AAPL", period="1y", interval="1d")
y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
pred = wrapper.predict(
    df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
    x_timestamp=df["timestamps"].tail(400),
    y_timestamp=pd.Series(y_ts, name="timestamps"),
    pred_len=30,
)
kronos_low  = float(pred["low"].min())
kronos_high = float(pred["high"].max())
kronos_range_pct = (kronos_high - kronos_low) / float(df["close"].iloc[-1]) * 100
print(f"Kronos predicted 30d range: {kronos_low:.2f} - {kronos_high:.2f} "
      f"({kronos_range_pct:.1f}% wide)")

# 2. Get the options market's implied vol (ATM 30-day call)
tkr = yf.Ticker("AAPL")
chain = tkr.option_chain()                      # nearest expiry chain
# Find the strike nearest to last close
last_close = float(df["close"].iloc[-1])
calls = chain.calls
nearest_idx = (calls["strike"] - last_close).abs().idxmin()
iv_pct = float(calls.loc[nearest_idx, "impliedVolatility"]) * 100
print(f"Options market implied vol:  {iv_pct:.1f}% "
      f"({calls.loc[nearest_idx, 'strike']} strike)")

# 3. Compare
if kronos_range_pct < iv_pct * 0.7:
    print(">> KRONOS NARROWER THAN MARKET — premium is overpriced.")
    print(">> STRATEGY: sell a covered call (or cash-secured put).")
elif kronos_range_pct > iv_pct * 1.3:
    print(">> KRONOS WIDER THAN MARKET — market is underpricing risk.")
    print(">> STRATEGY: buy protective put (or stay flat / avoid selling premium).")
else:
    print(">> Roughly aligned — no edge, sit out.")

Risk note: Selling options has unlimited downside on uncovered naked calls. Only sell covered calls (against shares you own) or cash-secured puts (against cash equal to strike × 100). Never naked.

4.3 Pairs / stat-arb trading

What it does: Forecast two correlated instruments, watch the spread between them, and trade when the spread's predicted move diverges from its realized move.

Why it works: Two correlated names usually mean-revert. When the spread drifts, it's a temporary dislocation. Kronos forecasts the direction of the reversion; you trade the magnitude.

Install:

pip install kronos-finance[global]
pip install statsmodels                  # for ADF test (mean-reversion proof)
pip install scikit-learn                 # for rolling correlation

Run:

import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import adfuller
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

PAIR_A, PAIR_B = "KO", "PEP"             # or "BTC/USDT", "ETH/USDT"
HORIZON = 30

# 1. Confirm the pair actually mean-reverts (skip if not)
a = load_ohlcv(PAIR_A, period="2y", interval="1d")
b = load_ohlcv(PAIR_B, period="2y", interval="1d")
spread = a["close"].values - b["close"].values
adf_p = adfuller(spread, maxlag=5)[1]
print(f"ADF p-value on spread: {adf_p:.3f} "
      f"(<0.05 = mean-reverting, good candidate)")

if adf_p >= 0.05:
    print("Spread is NOT mean-reverting. Choose another pair.")
else:
    # 2. Forecast each leg separately
    wrapper = load_kronos(model_id="small", device="cpu")
    def forecast_one(ticker):
        d = load_ohlcv(ticker, period="1y", interval="1d")
        last = d["timestamps"].iloc[-1]
        y_ts = pd.date_range(last, periods=HORIZON + 1, freq="1D")[1:]
        return wrapper.predict(
            df=d[["open", "high", "low", "close", "volume", "amount"]].tail(400),
            x_timestamp=d["timestamps"].tail(400),
            y_timestamp=pd.Series(y_ts, name="timestamps"),
            pred_len=HORIZON,
        )
    pred_a, pred_b = forecast_one(PAIR_A), forecast_one(PAIR_B)

    # 3. Predicted spread move
    pred_spread_t0 = float(pred_a["close"].iloc[0] - pred_b["close"].iloc[0])
    pred_spread_tH = float(pred_a["close"].iloc[-1] - pred_b["close"].iloc[-1])
    pred_delta = pred_spread_tH - pred_spread_t0
    print(f"\nPredicted spread change over {HORIZON}d: {pred_delta:+.2f}")

    # 4. Decide side
    last_spread = float(a["close"].iloc[-1] - b["close"].iloc[-1])
    z_score = (last_spread - spread.mean()) / spread.std()
    print(f"Current z-score: {z_score:+.2f}")

    # Simple rule: if z>1 and Kronos thinks spread falls, long B/short A
    #              if z<-1 and Kronos thinks spread rises, long A/short B
    if z_score > 1.0 and pred_delta < 0:
        action = f"LONG {PAIR_B} / SHORT {PAIR_A}"
    elif z_score < -1.0 and pred_delta > 0:
        action = f"LONG {PAIR_A} / SHORT {PAIR_B}"
    else:
        action = "HOLD (no mean-reversion signal)"
    print(f">> ACTION: {action}")

Risk note: Pairs fail when the correlation breaks (one company merges, scandal, delisting). Always have an exit rule: if the rolling 60-day correlation drops below 0.5, unwind the pair.

4.4 Event-driven reaction

What it does: Establishes a "no-event" forecast, lets the event happen (earnings, Fed, CPI), then runs a second forecast using the post-event price action. The gap between the two is the "surprise" you can trade.

Why it works: Markets move on the gap between expectation and reality. Kronos's pre-event forecast IS the expectation model; the post-event model captures the new reality. The difference is the trade.

Install:

pip install kronos-finance[global]
pip install pandas_market_calendars       # NYSE / earnings calendar

Run (manual event workflow):

import json
from datetime import datetime
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

# 1. BEFORE the event: capture the baseline forecast
EVENT_DATE = "2026-10-30"               # edit to your event
TICKER = "AAPL"

df = load_ohlcv(TICKER, period="1y", interval="1d")
last_ts_before = df["timestamps"].iloc[-1]
y_ts = pd.date_range(last_ts_before, periods=31, freq="1D")[1:]

wrapper = load_kronos(model_id="small", device="cpu")
baseline = wrapper.predict(
    df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
    x_timestamp=df["timestamps"].tail(400),
    y_timestamp=pd.Series(y_ts, name="timestamps"),
    pred_len=30,
)
baseline_record = {
    "ticker": TICKER,
    "event": "earnings",                # edit
    "captured_at": datetime.utcnow().isoformat(),
    "predicted_30d_close": float(baseline["close"].iloc[-1]),
}
with open(f"baseline_{TICKER}_{EVENT_DATE}.json", "w") as f:
    json.dump(baseline_record, f, indent=2)
print(f"Saved baseline to baseline_{TICKER}_{EVENT_DATE}.json")
print(f"  Pre-event predicted 30d close: ${baseline_record['predicted_30d_close']:.2f}")

# 2. WAIT FOR THE EVENT. Then run the same script with the same args.
#    Kronos will see new price action and produce a NEW forecast.
# 3. Compare the two:
#    new_record = {...}
#    surprise_pct = (new_record["predicted_30d_close"]
#                     - baseline_record["predicted_30d_close"]) / last * 100
#    if surprise_pct > +5%: market is more bullish than expected -> LONG
#    if surprise_pct < -5%: market is more bearish -> SHORT or reduce

For an automated approach, use pandas_market_calendars to know when US equities are closed for events, then run this script daily.

4.5 Crypto funding-rate arbitrage

What it does: Perpetual futures charge "funding" every 8 hours — longs pay shorts when funding is positive (the contract is priced above spot). Kronos's predicted direction can confirm or contradict the funding-implied direction. When they disagree, you trade.

Why it works: Funding rates embed market sentiment (longs are willing to pay shorts for leverage). Kronos gives an independent direction signal. Combining the two lets you fade extreme funding when Kronos disagrees, with the funding payment compensating you while you wait.

Install:

pip install kronos-finance[crypto]      # installs ccxt

Run:

import ccxt
import pandas as pd
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

SYMBOL = "BTC/USDT"
exchange = ccxt.binance({"enableRateLimit": True})

# 1. Get current funding rate (positive = longs pay shorts)
funding = exchange.fetch_funding_rate(SYMBOL)
rate_pct = float(funding["fundingRate"]) * 100
print(f"Current funding rate: {rate_pct:+.4f}% per 8h "
      f"({rate_pct * 3 * 365:+.1f}% annualized)")

# 2. Get Kronos's predicted direction
wrapper = load_kronos(model_id="small", device="cpu")
df = load_ohlcv(SYMBOL, source="ccxt", exchange="binance",
                interval="5m", limit=500)
y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=121, freq="5min")[1:]
pred = wrapper.predict(
    df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
    x_timestamp=df["timestamps"].tail(400),
    y_timestamp=pd.Series(y_ts, name="timestamps"),
    pred_len=120, T=1.0, top_p=0.9,
)
last_close = float(df["close"].iloc[-1])
expected = (float(pred["close"].iloc[-1]) - last_close) / last_close * 100
print(f"Kronos expected 10h move: {expected:+.2f}%")

# 3. Decision matrix
if rate_pct > 0.01 and expected < -0.5:
    print(">> Funding is POSITIVE (longs over-leveraged), Kronos is BEARISH.")
    print(">> ACTION: SHORT the perp, LONG spot to delta-hedge.")
    print("            Collect funding every 8h until Kronos turns bullish.")
elif rate_pct < -0.01 and expected > 0.5:
    print(">> Funding is NEGATIVE (shorts over-leveraged), Kronos is BULLISH.")
    print(">> ACTION: LONG the perp, SHORT spot to delta-hedge.")
    print("            Collect funding until Kronos turns bearish.")
else:
    print(">> No edge: funding and Kronos agree. Skip.")

Risk note: Funding-rate arb involves both legs moving. If the perp dumps and the spot doesn't, your hedge ratio drifts. Re-hedge every hour or use a market-neutral rebalancer. Also: exchange fees on the spot leg eat into the funding collection.

4.6 Slack / Discord / Telegram signal bot

What it does: Posts daily forecast summaries to a private chat channel. Read-only — no trading. Useful as a "second opinion" or as a notification system before you trade manually.

Why it works: Decisions made in isolation are worse than decisions made with a daily nudge. Even a 30-line bot changes behavior because it forces the model output into a human-readable summary at a consistent time each day.

Install:

pip install kronos-finance[global]
pip install slack-sdk                    # for Slack
pip install discord.py                   # for Discord
pip install python-telegram-bot          # for Telegram

Slack example (read the Slack docs on webhooks first):

# 1. Create an Incoming Webhook in your Slack workspace.
#    Settings -> Apps -> Incoming Webhooks -> Add to Slack.
#    Copy the webhook URL.
import os
from slack_sdk.webhook import WebhookClient
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

slack_url = os.environ["SLACK_WEBHOOK_URL"]   # never commit this
client = WebhookClient(slack_url)

WATCHLIST = ["AAPL", "MSFT", "NVDA"]
wrapper = load_kronos(model_id="small", device="cpu")

def fmt(ticker):
    df = load_ohlcv(ticker, period="1y", interval="1d")
    y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
        x_timestamp=df["timestamps"].tail(400),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=30,
    )
    last = float(df["close"].iloc[-1])
    p30 = float(pred["close"].iloc[-1])
    ret = (p30 - last) / last * 100
    arrow = "🟢" if ret > 0 else "🔴"
    return f"{arrow} *{ticker}*: ${last:.2f} → ${p30:.2f} ({ret:+.2f}% in 30d)"

message = "*Kronos daily forecast*\n" + "\n".join(fmt(t) for t in WATCHLIST)
client.send_text(text=message)

The cron entry (Linux/Mac):

0 8 * * 1-5  cd /home/user/kronos-finance && \
            /usr/bin/python3 daily_signal_post.py

For Windows Task Scheduler: same pattern, point at python.exe and daily_signal_post.py. For GitHub Actions schedule: see examples/19_cronjob_daily_forecast.py for the lockfile + JSONL-log pattern that translates directly.

Bot safety: never put trade execution logic in the same bot as the Slack poster. Run them in separate processes with separate credentials. If the Slack token leaks, the worst case is "spammer posts bad forecasts to your private channel."

4.7 Risk dashboard for held positions

What it does: Runs Kronos against every position you currently hold and alerts if the predicted 30-day drawdown exceeds your pre-set loss tolerance. No trading — pure early warning.

Why it works: Most losses come from holding a position through a regime change. Kronos sees the regime shift in price + volume before the news does. The signal isn't "sell" — it's "re-evaluate this position."

Install:

pip install kronos-finance[global]

Run:

# positions.csv format: ticker,shares,cost_basis
# AAPL,100,150.00
# NVDA,50,400.00
import pandas as pd
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

positions = pd.read_csv("positions.csv")
LOSS_TOLERANCE_PCT = 15                    # alert if predicted drawdown > this
wrapper = load_kronos(model_id="small", device="cpu")

alerts = []
for _, row in positions.iterrows():
    ticker, shares, cost = row["ticker"], int(row["shares"]), float(row["cost_basis"])
    df = load_ohlcv(ticker, period="1y", interval="1d")
    y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
        x_timestamp=df["timestamps"].tail(400),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=30,
    )
    worst_predicted = float(pred["low"].min())
    worst_drawdown_pct = (worst_predicted - cost) / cost * 100
    current_pnl = (float(df["close"].iloc[-1]) - cost) / cost * 100

    if worst_drawdown_pct < -LOSS_TOLERANCE_PCT:
        alerts.append({
            "ticker": ticker,
            "shares": shares,
            "current_pnl_pct": round(current_pnl, 2),
            "predicted_30d_worst_drawdown_pct": round(worst_drawdown_pct, 2),
            "action": "REVIEW POSITION",
        })
    else:
        print(f"  {ticker:6s}: current P&L {current_pnl:+.2f}%, "
              f"predicted worst 30d drawdown {worst_drawdown_pct:+.2f}% — OK")

if alerts:
    print("\n!!! POSITIONS NEED REVIEW !!!")
    for a in alerts:
        print(f"  {a['ticker']}: {a['predicted_30d_worst_drawdown_pct']:+.2f}% "
              f"predicted drawdown vs {a['current_pnl_pct']:+.2f}% current P&L")
else:
    print("\nAll positions within predicted risk tolerance.")

Pipe the alerts list to your Slack webhook from 4.6 above for a real-time risk dashboard.

4.8 Macro overlay / market-timing filter

What it does: Runs Kronos once a day on broad indexes (SPY, QQQ, BTC, gold ETF). When the aggregate expected return across all of them is negative, raise cash. When positive, be fully invested.

Why it works: Markets cluster — when broad indexes sell off, most stocks sell off with them. A single-market signal can shield a long-only portfolio from the worst drawdowns. It's not fancy, it's not high-Sharpe, but it's historically robust and easy to implement.

Install:

pip install kronos-finance[global]

Run:

from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

UNIVERSE = ["SPY", "QQQ", "IWM", "GLD"]    # broad-market ETFs
THRESHOLD_PCT = 1.0                       # raise cash if aggregate < +1%
wrapper = load_kronos(model_id="small", device="cpu")

expected_returns = []
for ticker in UNIVERSE:
    df = load_ohlcv(ticker, period="1y", interval="1d")
    y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
    pred = wrapper.predict(
        df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
        x_timestamp=df["timestamps"].tail(400),
        y_timestamp=pd.Series(y_ts, name="timestamps"),
        pred_len=30,
    )
    last = float(df["close"].iloc[-1])
    p30 = float(pred["close"].iloc[-1])
    expected_returns.append((p30 - last) / last * 100)

aggregate = sum(expected_returns) / len(expected_returns)
print(f"Aggregate 30d expected return across {UNIVERSE}: {aggregate:+.2f}%")

if aggregate > THRESHOLD_PCT:
    print(">> MACRO SIGNAL: bullish. Be long.")
elif aggregate < -THRESHOLD_PCT:
    print(">> MACRO SIGNAL: bearish. Raise cash or hedge.")
else:
    print(">> MACRO SIGNAL: neutral. No action.")

Run this daily and act on the signal. Add a "no-trade zone" (aggregate in (-1%, +1%)) to avoid over-trading.

4.9 News corroboration with FinBERT

What it does: Pulls today's news for a ticker, runs sentiment analysis with FinBERT, and combines it with Kronos's expected return. When both agree, conviction is high. When they disagree, the model output is suspect.

Why it works: Models fail differently. Kronos fails on regime changes it hasn't seen; FinBERT fails on sarcasm or unusual phrasing. Two independent signals are more reliable than one, especially when the signal disagrees.

Install:

pip install kronos-finance[global]
pip install transformers torch
pip install newsapi-python                # news API client
# export NEWSAPI_KEY=...

Run:

import os
import pandas as pd
from transformers import pipeline
from newsapi import NewsApiClient
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

newsapi = NewsApiClient(api_key=os.environ["NEWSAPI_KEY"])
finbert = pipeline("sentiment-analysis",
                   model="ProsusAI/finbert",
                   tokenizer="ProsusAI/finbert")

TICKER = "AAPL"
QUERY = TICKER if TICKER != "BTC/USDT" else "bitcoin"

# 1. News sentiment
articles = newsapi.get_everything(q=QUERY, language="en",
                                  page_size=20, sort_by="relevancy")
scores = []
for a in articles["articles"]:
    text = (a["title"] or "") + ". " + (a["description"] or "")
    if text.strip():
        r = finbert(text[:512])[0]                  # FinBERT caps at 512 tokens
        # Map to signed score: +1 positive, -1 negative, 0 neutral
        sign = 1.0 if r["label"] == "positive" else (-1.0 if r["label"] == "negative" else 0.0)
        scores.append(sign * r["score"])
news_signal = sum(scores) / len(scores) if scores else 0.0
print(f"FinBERT news signal ({len(scores)} articles): {news_signal:+.3f}")

# 2. Kronos signal
wrapper = load_kronos(model_id="small", device="cpu")
df = load_ohlcv(TICKER, period="1y", interval="1d")
y_ts = pd.date_range(df["timestamps"].iloc[-1], periods=31, freq="1D")[1:]
pred = wrapper.predict(
    df=df[["open", "high", "low", "close", "volume", "amount"]].tail(400),
    x_timestamp=df["timestamps"].tail(400),
    y_timestamp=pd.Series(y_ts, name="timestamps"),
    pred_len=30,
)
last = float(df["close"].iloc[-1])
p30 = float(pred["close"].iloc[-1])
kronos_signal = (p30 - last) / last * 100
print(f"Kronos 30d signal: {kronos_signal:+.2f}%")

# 3. Combine
def bucket(x, bull=2.0, bear=-2.0):
    if x > bull:  return "BULLISH"
    if x < bear:  return "BEARISH"
    return "NEUTRAL"

k = bucket(kronos_signal)
n = bucket(news_signal * 100)              # scale FinBERT to comparable magnitude
print(f"\nKronos: {k}  |  News: {n}")

if k == n and k != "NEUTRAL":
    print(">>> BOTH AGREE — HIGH CONVICTION trade.")
elif k != "NEUTRAL" and n != "NEUTRAL" and k != n:
    print(">>> CONFLICT — investigate before trading (regime change likely).")
else:
    print(">>> No actionable signal.")

Risk note: News APIs have rate limits. newsapi free tier caps at 100 requests/day; for production use, cache the day's articles and only re-query on ticker change.

4.10 Multi-signal ensemble

What it does: Combines Kronos with every other signal in this section into a single weighted score per ticker per day. The weighting is the art — start with equal weights and let walk-forward validation tune them.

Why it works: Ensemble methods beat single models in nearly every domain, finance included. Kronos is one of many inputs; the ensemble averages out the per-model failure modes.

Install:

pip install kronos-finance[global,analysis]

Run:

# Wire all the signals from sections 4.1 - 4.9 into a single daily score.
# The pattern is identical each time:
#   1. Run the signal logic on the same ticker / same day.
#   2. Map the output to a signed number in [-1, +1].
#   3. Multiply by a weight; sum across signals.
#   4. Trade when the weighted sum crosses a threshold.

import pandas as pd
from kronos_finance import load_kronos
from kronos_finance.data import load_ohlcv

WEIGHTS = {
    "kronos_direction":    0.30,    # 4.1 / standard signal
    "kronos_vol_sizing":   0.15,    # 4.1 / Sharpe-like ratio
    "options_iv_edge":     0.15,    # 4.2
    "pairs_signal":        0.10,    # 4.3
    "macro_overlay":       0.15,    # 4.8
    "news_sentiment":      0.15,    # 4.9
}
TICKER = "AAPL"

# Compute each component (pseudo-code — wire the sections above):
components = {
    "kronos_direction":    0.40,    # +0.40 = +4% expected return normalized
    "kronos_vol_sizing":   0.20,
    "options_iv_edge":     -0.30,   # market is pricing more vol than Kronos
    "pairs_signal":        0.50,
    "macro_overlay":       0.60,
    "news_sentiment":      0.20,
}
ensemble_score = sum(WEIGHTS[k] * v for k, v in components.items())
print(f"Ensemble score for {TICKER}: {ensemble_score:+.3f}")

if ensemble_score > 0.30:
    print(">>> Strong ensemble signal: consider long.")
elif ensemble_score < -0.30:
    print(">>> Strong ensemble signal: consider short / reduce.")
else:
    print(">>> Weak or mixed: hold current position.")

Use examples/06_indicators_and_tearsheet.py's quantstats HTML output to compare the ensemble's performance against any single signal over the same window. The ensemble almost always wins on risk-adjusted return, but check anyway.

5. Live automation patterns

A daily forecast that fires automatically (cron / Task Scheduler) is the most common production pattern. See examples/19_cronjob_daily_forecast.py for the basics. To go further:

• Multi-ticker fan-out with retry. Wrap each forecast in try/except + exponential backoff. One bad ticker (rate-limited, delisted) should not stop the rest.
• Slack/email on anomaly. If a forecast's expected return moves

  3% from yesterday's prediction, send an alert. The model's change is often more informative than its level.

• Health checks before orders. Run the health-check example (examples/17_healthcheck_environment.py) as a precondition. If CUDA is broken or HF cache is corrupt, halt the auto-trader.
• Audit log every order. Persist every order decision (input DataFrame, prediction, signal logic, sizing, timestamp, account state) to a JSONL file. When (not if) you need to debug a bad day, the audit log is the only way.
• Daily-loss circuit breaker. Track cumulative P&L for the day; if it's worse than -X% of starting equity, halt the bot for the rest of the day. This is more important than the strategy itself.
• Idempotency keys. Network retries can submit duplicate orders. Use idempotency keys (Alpaca, IBKR both support them) keyed on (ticker, date, side, qty).
• Read-only mode first. For the first 30 days, run the bot with execution disabled — log what it would have done. Compare against your paper broker's actual fill prices. Only enable execution when the read-only logs match the broker.

6. Tools that pair well with Kronos

You don't need to write everything from scratch. These composable pieces each fill a gap.

• vectorbt.pro — paid, but the only Python backtester that runs a 10-year tick-level crypto strategy in seconds. Worth it for serious sweeps.
• quantstats — already a dep of [analysis]. Pull-lev, factor analysis, HTML tearsheets. Use it to compare strategy variants.
• riskfolio-lib — portfolio optimization (HRP, mean-variance, Black-Litterman). Combine Kronos expected returns with riskfolio-lib's covariance model for a complete portfolio construction stack.
• empyrical (deprecated) / pyfolio-reloaded — risk and performance metrics.
• pandas-ta — already a dep. ~130 indicators. Don't reinvent RSI / MACD / Bollinger / ATR / OBV.
• yfinance — already a dep. Free OHLCV for US/CN/HK tickers with no API key.
• akshare — already a dep. CN A-share data.
• ccxt — already a dep. 100+ crypto exchanges.
• alpaca-trade-api / ib_insync / binance — broker APIs as listed above.
• great-expectations — schema validation on input data. Catches "today's data has NaN, the model silently predicted NaN, and the bot bought 10,000 shares of garbage." Cheap insurance.
• pydantic — already a dep. Validate signal outputs before they hit the broker.
• apscheduler / croniter — robust scheduling if you don't want raw cron.

7. What "good" looks like

After 6 months of running a Kronos-based strategy on a paper account, expect:

• Win rate: 50-60% on daily-bar direction. Higher than random but lower than naive backtests suggest.
• Sharpe ratio: 0.5-1.5 after costs, if you've tuned the signal logic well. Realistic, not glamorous.
• Max drawdown: 10-25% on the underlying, regardless of strategy. The model doesn't prevent drawdowns; it filters which direction you take them.
• Turnover: 1-5 trades/day across a 10-ticker watchlist if you retrain daily. Costs matter: even $0.005/share adds up at 1k shares/day.
• Drift: the model's accuracy will slowly degrade over months as market regime shifts. Plan to re-evaluate the signal logic quarterly.

If your backtest claims >2.0 Sharpe, >70% win rate, and <5% drawdown on daily bars, it has a bug. Go find it.

8. Where to learn more

• examples/05_batch_predict.py — multi-ticker sweep
• examples/06_indicators_and_tearsheet.py — indicators + quantstats
• examples/14_quantile_bands.py — distribution-aware sizing
• examples/19_cronjob_daily_forecast.py — daily automation
• examples/21_portfolio_kronos_weighting.py — three weighting schemes
• Backtesting frameworks: see the table above; start with backtesting.py for sanity checks, then vectorbt for sweeps, then backtrader for realism.
• Live execution: start with Alpaca paper trading (free API key, 5-minute setup) before touching any real-money broker.

And again: this is research software. Don't risk what you can't afford to lose. The first version of your strategy should run in paper mode for at least 30 trading days before you ever let it touch a live account.

Troubleshooting

The most common pitfalls — full list in docs/TROUBLESHOOTING.md:

• error: externally-managed-environment — you're trying to pip install into system Python. Use a venv (see the section at the top of this README).
• Model download stalls / 401 from HuggingFace — set HF_TOKEN or run huggingface-cli login.
• CUDA version mismatch — install PyTorch from the matching CUDA index URL.
• yfinance rate limit — switch to auto source or add period= to limit.
• AKShare returns empty — the ticker may have delisted. Try yfinance with 600519.SS.
• Playwright browser missing — playwright install --with-deps chromium.
• max_context exceeded — reduce lookback or use Kronos-mini (2048 context).
• Tz-aware timestamp warning — convert to UTC and drop the tz before passing in.
• PermissionError on Windows — run your terminal as Administrator or use a venv.

Hardware (CPU vs CUDA)

Kronos runs on both CPU and CUDA GPUs. The default pip install kronos-finance installs the standard PyPI torch wheel, which is CUDA-enabled by default and works on either. There's no separate "GPU version" of kronos-finance.

You pick the device at runtime via --device (CLI) or device= (Python):

kronos predict AAPL --device cpu       # default; works everywhere
kronos predict AAPL --device cuda     # first GPU
kronos predict AAPL --device cuda:0   # specific GPU

wrapper = load_kronos(model_id="small", device="cuda")

Which device should I use?

Model	Params	CPU latency (30-step forecast)	CUDA latency	Recommendation
Kronos-mini	4.1M	~2s	~0.2s	Either works
Kronos-small	24.7M	~5s	~0.3s	Either works
Kronos-base	102.3M	~15s	~1s	CUDA recommended

Memory

• CPU: ~1 GB RAM for small, +500 MB for base.
• CUDA: ~1 GB VRAM for small, ~2 GB VRAM for base. An RTX 3060 (12 GB) is plenty.

Pinning your CUDA version

If you have an NVIDIA GPU, install PyTorch from the matching CUDA index URL before installing kronos-finance to control which CUDA toolkit version is bundled:

# CUDA 12.1 — match your NVIDIA driver
pip install torch --index-url https://download.pytorch.org/whl/cu121
pip install kronos-finance[global]

# CUDA 11.8
pip install torch --index-url https://download.pytorch.org/whl/cu118
pip install kronos-finance[global]

# CPU-only (smaller download, ~200 MB instead of ~800 MB)
pip install torch --index-url https://download.pytorch.org/whl/cpu
pip install kronos-finance[global]

Apple Silicon (M1/M2/M3)

pip install kronos-finance on Apple Silicon uses PyTorch's MPS backend automatically. Pass --device mps in the CLI or device="mps" in the API.

Common CUDA errors

• CUDA error: no kernel image is available — your PyTorch CUDA version doesn't match your NVIDIA driver. Reinstall PyTorch from the matching index URL above.
• CUDA out of memory — your GPU is too small. Use Kronos-mini or reduce lookback. On CPU there's no such limit (just slower).
• CUDA unavailable but requested — your install doesn't have CUDA support, or no GPU is visible. nvidia-smi to check.

How it works

Ticker (e.g. "AAPL")
    │
    ▼  load_ohlcv() auto-detects source -> yfinance
pd.DataFrame [timestamps, open, high, low, close, volume, amount]
    │
    ▼  load_kronos() downloads Kronos-small from HuggingFace
KronosWrapper (model + tokenizer + predictor)
    │
    ▼  wrapper.predict() runs autoregressive Transformer
pd.DataFrame [predicted OHLCV]
    │
    ▼  analysis: indicators + backtest -> HTML tearsheet

Performance & limits

Setting	Default	Cap	Note
pred_len	30	1000 (CLI), no cap in API	Larger = slower inference
lookback	400	512 (small/base/large), 2048 (mini)	Auto-truncated
sample_count	1	20	Quantile bands need >1
Memory (CPU, small)	~1GB	—	+500MB for base
Memory (CUDA, base)	~2GB VRAM	—	RTX 3060+ recommended
Latency (CPU, 30-step, 1 ticker)	~5s	—	~1s on CUDA

Development

git clone https://github.com/lordxmen2k/kronos-finance.git
cd kronos-finance
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev,test,ui]"
pytest                         # unit + CLI tests
ruff check src/                # lint

See docs/CONTRIBUTING.md.

Citation

If you use this in research, please cite the original Kronos paper:

@inproceedings{kronos2026,
  title  = {Kronos: A Foundation Model for the Language of Financial Markets},
  author = {Shi, Yu and others},
  booktitle = {AAAI},
  year   = {2026},
}

And this wrapper:

@software{kronos_finance,
  author = {lordxmen2k},
  title  = {kronos-finance: A Python wrapper for Kronos},
  year   = {2026},
  url    = {https://github.com/lordxmen2k/kronos-finance}
}

License

MIT — see LICENSE for the full text. The original Kronos project is also MIT; see src/kronos_finance/_vendor/LICENSE_KRONOS for the vendored upstream license.

Acknowledgements

• The NeoQuasar team for the Kronos foundation model.
• HuggingFace for model hosting.
• AKShare, yfinance, CCXT, Qlib — the data layer.
• pandas-ta, quantstats — analysis.
• Plotly, Flask — the dashboard.

Glossary

See docs/INSTALL.md#glossary for the full 35-term glossary. A quick index of the most important ones:

• OHLCV — Open, High, Low, Close, Volume. The five columns Kronos expects.
• K-line — Chinese term for candlestick; same thing.
• Context length / max_context — maximum past bars the model can see (512 for small/base/large; 2048 for mini).
• AR / autoregressive — generates outputs one step at a time.
• Tokenizer — converts continuous OHLCV to discrete tokens before the Transformer.
• Sample count — number of forecast paths to draw; more = smoother quantile band.
• Tearsheet — one-page performance report; quantstats generates HTML.
• Nucleus sampling (top-p) — sampling from smallest token set whose cumulative prob ≥ p.
• Temperature (T) — sampling temperature; T<1 conservative, T>1 exploratory.
• Quantile band — uncertainty interval drawn when sample_count > 1.
• PEP 668 — Python spec marking system Python as externally managed; use venv.
• Twine — twine upload dist/* publishes to PyPI.
• Wheel (.whl) — built distribution format.