reader.py

"""Read OctoBot backtest export (run_data, portfolio_value, trades); single run or metadata-based scan."""
from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any


def load_json(path: Path) -> dict[str, Any] | None:
    if not path.is_file():
        return None
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except json.JSONDecodeError:
        return None
    except PermissionError:
        print(f"[reader] Permission denied: {path}", file=sys.stderr)
        return None
    except OSError:
        return None


def find_exchange_dirs(run_dir: Path) -> list[Path]:
    if not run_dir.is_dir():
        return []
    return [d for d in run_dir.iterdir() if d.is_dir() and not d.is_symlink() and not d.name.startswith(".")]


def get_run_data(run_dir: Path) -> dict[str, Any] | None:
    return load_json(run_dir / "run_data.json")


def get_metadata_entry(data: dict[str, Any], run_id: str | None = None) -> dict[str, Any] | None:
    """Return metadata for run_id (e.g. '3' for backtesting_3). Fallback to '1'/'0'/first if run_id missing."""
    meta = data.get("metadata")
    if not meta or not isinstance(meta, dict):
        return None
    if run_id and isinstance(meta.get(run_id), dict):
        return meta[run_id]
    for k in ("1", "0"):
        if k in meta and isinstance(meta[k], dict):
            return meta[k]
    return next((v for v in meta.values() if isinstance(v, dict)), None)


def get_inputs_for_params(data: dict[str, Any]) -> dict[str, Any]:
    inputs = data.get("inputs")
    if not inputs or not isinstance(inputs, dict):
        return {}
    strategy_name = "OctoBot"
    for inp in inputs.values():
        if isinstance(inp, dict) and inp.get("name") in ("strategy", "trading_mode") and inp.get("value"):
            strategy_name = str(inp["value"])
            break
    params: dict[str, Any] = {"strategy": strategy_name}
    reserved = {"strategy"}
    for key, inp in sorted(inputs.items(), key=lambda x: (len(str(x[0])), str(x[0]))):
        if not isinstance(inp, dict):
            continue
        name = inp.get("name")
        value = inp.get("value")
        if name in ("pair_settings", "pair") and value is not None:
            if name == "pair" and isinstance(value, str):
                params["pair"] = value
            elif name == "pair_settings" and isinstance(value, list) and value:
                first = value[0]
                if isinstance(first, dict):
                    for k, v in first.items():
                        if k in reserved or (k == "pair" and params.get("pair")):
                            continue
                        if k not in params:
                            params[k] = v
        elif name and value is not None and name not in ("pair_settings",) and name not in reserved:
            params[name] = value
    return params


def get_portfolio_value(exchange_dir: Path) -> dict[str, Any] | None:
    files = sorted(exchange_dir.glob("portfolio_value_*.json"), key=lambda f: f.name)
    return load_json(files[-1]) if files else None


def get_trades_file(exchange_dir: Path) -> dict[str, Any] | None:
    files = sorted(exchange_dir.glob("trades_*.json"), key=lambda f: f.name)
    return load_json(files[-1]) if files else None


def get_all_trades_list(trades_data: dict[str, Any]) -> list[dict[str, Any]]:
    all_trades = trades_data.get("all_trades")
    if not all_trades or not isinstance(all_trades, dict):
        return []
    out = []
    for k in sorted(all_trades.keys(), key=lambda x: int(x) if x.isdigit() else 0):
        v = all_trades[k]
        if not isinstance(v, dict):
            continue
        base = v.get("origin_value")
        merged = {**v, **(base if isinstance(base, dict) else {})}
        out.append(merged)
    return out


def get_historical_portfolio_value(
    portfolio_data: dict[str, Any],
    ref_market: str = "USDC",
) -> list[tuple[int, float]]:
    hist = portfolio_data.get("historical_portfolio_value")
    if not hist or not isinstance(hist, dict):
        return []
    out: list[tuple[int, float]] = []
    for k in sorted(hist.keys(), key=lambda x: int(x) if x.isdigit() else 0):
        v = hist.get(k)
        if not isinstance(v, dict):
            continue
        t = v.get("t")
        vv = v.get("v")
        if not isinstance(vv, dict) or t is None:
            continue
        val = vv.get(ref_market)
        if not isinstance(val, (int, float)):
            val = next((x for x in vv.values() if isinstance(x, (int, float))), None)
        if val is not None:
            t_val = int(t) if isinstance(t, (int, float)) else 0
            t_sec = t_val // 1000 if t_val > 1e11 else t_val
            out.append((t_sec, float(val)))
    return out


def collect_run(run_dir: Path) -> dict[str, Any] | None:
    """Collect one run: run_data, metadata, parameters, trades_list, equity_series. None if incomplete."""
    run_data = get_run_data(run_dir)
    if not run_data:
        return None
    raw_id = run_dir.name.replace("backtesting_", "").strip()
    run_id = raw_id if raw_id.isdigit() else None
    meta = get_metadata_entry(run_data, run_id)
    if not meta:
        return None
    exchange_dirs = find_exchange_dirs(run_dir)
    exchange_dir = None
    portfolio_data = None
    trades_data = None
    for d in exchange_dirs:
        pv = get_portfolio_value(d)
        tr = get_trades_file(d)
        if pv and tr:
            exchange_dir = d
            portfolio_data = pv
            trades_data = tr
            if len(exchange_dirs) > 1:
                print("[reader] Multiple exchange dirs found, using first with both portfolio_value and trades.", file=sys.stderr)
            break
    if not exchange_dir or not trades_data:
        return None
    trades_list = get_all_trades_list(trades_data)
    ref_market = (meta.get("ref_market") or "USDC") if isinstance(meta.get("ref_market"), str) else "USDC"
    equity_series = (
        get_historical_portfolio_value(portfolio_data, ref_market) if portfolio_data else []
    )
    return {
        "run_data": run_data,
        "metadata": meta,
        "parameters": get_inputs_for_params(run_data),
        "exchange": exchange_dir.name.lower() if exchange_dir else None,
        "ref_market": ref_market,
        "trades_list": trades_list,
        "equity_series": equity_series,
    }


def collect_from_metadata(parent_dir: Path) -> list[tuple[str, Path]]:
    """List (run_id, backtesting_N path) from parent_dir/metadata.json. Only dirs with run_data.json."""
    meta_file = parent_dir / "metadata.json"
    data = load_json(meta_file)
    if not data:
        return []
    meta = data.get("metadata")
    if not meta or not isinstance(meta, dict):
        return []
    runs: list[tuple[str, Path]] = []
    for k, v in meta.items():
        if isinstance(v, dict) and (v.get("id") or v.get("timestamp")):
            folder_name = f"backtesting_{k}"
            path = parent_dir / folder_name
            if path.is_dir() and (path / "run_data.json").is_file():
                runs.append((k, path))

    ts_cache = {
        k: float(v.get("timestamp") or v.get("start_time") or 0)
        for k, v in meta.items()
        if isinstance(v, dict)
    }
    runs.sort(key=lambda item: (int(item[0]) if item[0].isdigit() else 0, ts_cache.get(item[0], 0)))
    return runs