`assets/bot-template/ledger.py`

116 lines
"""
ledger.py — append-only paper-trading ledger + position/P&L accounting.

Every fill is appended as one JSON line to data/ledger.jsonl (append-only = you
can always reconstruct state and never silently lose a trade). Open positions and
realized/unrealized P&L are derived by replaying the log — there is no mutable
"current state" file to corrupt.

Paper only. The live path (executor.py) is gated and does not write here until a
real broker fill is confirmed.
"""

import json
import os
from datetime import datetime, timezone

LEDGER_PATH = os.path.join(os.path.dirname(__file__), "data", "ledger.jsonl")


def _now():
    return datetime.now(timezone.utc).isoformat()


def _ensure_dir():
    os.makedirs(os.path.dirname(LEDGER_PATH), exist_ok=True)


def record_fill(market_id, title, side, action, price, size_usd, reason, mode):
    """Append one fill. action is 'open' or 'close'."""
    _ensure_dir()
    row = {
        "ts": _now(), "market_id": market_id, "title": title, "side": side,
        "action": action, "price": round(float(price), 4),
        "size_usd": round(float(size_usd), 2), "reason": reason, "mode": mode,
    }
    with open(LEDGER_PATH, "a") as f:
        f.write(json.dumps(row) + "\n")
    return row


def _read_all():
    if not os.path.exists(LEDGER_PATH):
        return []
    with open(LEDGER_PATH) as f:
        return [json.loads(line) for line in f if line.strip()]


def open_positions():
    """Replay the log → dict of market_id -> open position."""
    positions = {}
    for r in _read_all():
        mid = r["market_id"]
        if r["action"] == "open":
            positions[mid] = {
                "market_id": mid, "title": r["title"], "side": r["side"],
                "entry_price": r["price"], "size_usd": r["size_usd"],
                "shares": r["size_usd"] / r["price"] if r["price"] else 0.0,
                "opened": r["ts"],
            }
        elif r["action"] == "close":
            positions.pop(mid, None)
    return positions


def realized_pnl():
    """Sum realized P&L across all closed round-trips."""
    entries, pnl = {}, 0.0
    for r in _read_all():
        mid = r["market_id"]
        if r["action"] == "open":
            entries[mid] = r
        elif r["action"] == "close" and mid in entries:
            e = entries.pop(mid)
            shares = e["size_usd"] / e["price"] if e["price"] else 0.0
            pnl += shares * (r["price"] - e["price"])
    return round(pnl, 2)


def total_exposure_usd():
    return round(sum(p["size_usd"] for p in open_positions().values()), 2)


def pnl_today_usd():
    today = datetime.now(timezone.utc).date().isoformat()
    entries, pnl = {}, 0.0
    for r in _read_all():
        mid = r["market_id"]
        if r["action"] == "open":
            entries[mid] = r
        elif r["action"] == "close" and mid in entries:
            e = entries.pop(mid)
            if r["ts"][:10] == today:
                shares = e["size_usd"] / e["price"] if e["price"] else 0.0
                pnl += shares * (r["price"] - e["price"])
    return round(pnl, 2)


def summary(mark_fn=None):
    """Snapshot for the dashboard/CLI. mark_fn(market_id)->price gives unrealized."""
    pos = open_positions()
    unreal = 0.0
    for p in pos.values():
        mark = mark_fn(p["market_id"]) if mark_fn else None
        if mark is not None:
            unreal += p["shares"] * (mark - p["entry_price"])
            p["mark"] = round(mark, 4)
            p["unrealized_usd"] = round(p["shares"] * (mark - p["entry_price"]), 2)
    return {
        "open_positions": list(pos.values()),
        "open_count": len(pos),
        "exposure_usd": total_exposure_usd(),
        "realized_pnl_usd": realized_pnl(),
        "unrealized_pnl_usd": round(unreal, 2),
        "pnl_today_usd": pnl_today_usd(),
    }