| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051 |
- import pandas as pd
- from okx_codex_trader.research_metrics import equity_metrics, horizon_rows, trade_stats, worst_month
- def test_equity_metrics_calculates_return_drawdown_and_calmar() -> None:
- frame = pd.DataFrame(
- [
- {"ts": pd.Timestamp("2025-01-01", tz="UTC"), "equity": 100.0},
- {"ts": pd.Timestamp("2025-07-01", tz="UTC"), "equity": 80.0},
- {"ts": pd.Timestamp("2026-01-01", tz="UTC"), "equity": 120.0},
- ]
- )
- metrics = equity_metrics(frame, int(frame["ts"].iloc[0].timestamp() * 1000), int(frame["ts"].iloc[-1].timestamp() * 1000))
- assert round(metrics["net_total_return"], 6) == 0.2
- assert round(metrics["net_max_drawdown"], 6) == 0.2
- assert metrics["net_annualized_return"] > 0.19
- assert metrics["net_calmar"] > 0.9
- def test_horizon_rows_uses_cutoff_equity_when_history_exists() -> None:
- frame = pd.DataFrame(
- [
- {"ts": pd.Timestamp("2025-01-01", tz="UTC"), "equity": 100.0},
- {"ts": pd.Timestamp("2026-01-01", tz="UTC"), "equity": 150.0},
- {"ts": pd.Timestamp("2026-02-01", tz="UTC"), "equity": 120.0},
- ]
- )
- last_ts = int(pd.Timestamp("2026-02-01", tz="UTC").timestamp() * 1000)
- rows = horizon_rows(frame, last_ts, horizons=(("30d", pd.DateOffset(days=30)),))
- assert rows[0]["horizon"] == "30d"
- assert rows[0]["horizon_start"] == "2026-01-02 00:00"
- assert rows[0]["net_total_return"] < 0.0
- def test_trade_stats_and_worst_month() -> None:
- stats = trade_stats([{"return_pct": 2.0}, {"return_pct": -1.0}, {"return_pct": 4.0}])
- frame = pd.DataFrame(
- [
- {"ts": pd.Timestamp("2026-01-01", tz="UTC"), "equity": 100.0},
- {"ts": pd.Timestamp("2026-01-31", tz="UTC"), "equity": 120.0},
- {"ts": pd.Timestamp("2026-02-28", tz="UTC"), "equity": 90.0},
- ]
- )
- assert stats == {"avg_return_pct": 5.0 / 3.0, "payoff_ratio": 3.0, "profit_factor": 6.0}
- assert worst_month(frame) == ("2026-02", -0.25)
|