hermes-mcp/simulation/run_replay.py

from __future__ import annotations

from pathlib import Path
import argparse
import json
import sys

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
SIM_SRC = ROOT / "simulation" / "src"
for path in (str(SRC), str(SIM_SRC)):
    if path not in sys.path:
        sys.path.insert(0, path)

from hermes_sim.candles import load_candles_csv
from hermes_sim.harness import ReplayConfig, rows_to_jsonl, run_replay


def main() -> int:
    parser = argparse.ArgumentParser(description="Replay Hermes decisions from historic candles.")
    parser.add_argument("--candles", required=True, help="Path to a CSV with timestamp,open,high,low,close,volume columns")
    parser.add_argument("--symbol", default="XRPUSD")
    parser.add_argument("--base", default="XRP")
    parser.add_argument("--quote", default="USD")
    parser.add_argument("--account-id", default="sim-account")
    parser.add_argument("--fee-rate", type=float, default=0.004)
    parser.add_argument("--horizon-bars", type=int, default=30)
    parser.add_argument("--lookback-bars", type=int, default=2000, help="How many most-recent 1m candles to use for feature/regime computation")
    parser.add_argument("--progress-every", type=int, default=2000, help="Print progress every N replay rows")
    parser.add_argument("--base-balance", type=float, default=500.0)
    parser.add_argument("--quote-balance", type=float, default=500.0)
    parser.add_argument("--inventory-state", default="balanced")
    parser.add_argument("--rebalance-needed", action="store_true")
    parser.add_argument("--out", help="Optional JSONL output file")
    args = parser.parse_args()

    candles = load_candles_csv(args.candles)
    config = ReplayConfig(
        market_symbol=args.symbol,
        base_currency=args.base,
        quote_currency=args.quote,
        account_id=args.account_id,
        fee_rate=args.fee_rate,
        horizon_bars=args.horizon_bars,
        base_balance=args.base_balance,
        quote_balance=args.quote_balance,
        inventory_state=args.inventory_state,
        rebalance_needed=args.rebalance_needed,
    )
    rows = run_replay(candles=candles, config=config, lookback_bars=args.lookback_bars, progress_every=args.progress_every)
    output = rows_to_jsonl(rows)

    if args.out:
        Path(args.out).write_text(output + ("\n" if output else ""), encoding="utf-8")
    else:
        print(output)

    summary = {
        "candles": len(candles),
        "rows": len(rows),
        "avg_score": round(sum(r.score for r in rows) / len(rows), 4) if rows else 0.0,
    }
    print(json.dumps(summary, indent=2), file=sys.stderr)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())