#!/usr/bin/env python3
"""Audit dry-run/live execution quality from a Freqtrade sqlite database.

Two questions, matching docs/strategy-iterations.md 开放前沿 #3:

1. Stop slippage: for every closed trade that exited via a stop
   (stop_loss / stoploss_on_exchange / trailing_stop_loss), compare the
   recorded stop price with the actual close rate.  Positive slippage
   percentage = adverse fill (cost), negative = favourable.
2. Entry fill rate: limit entry orders that filled vs. were cancelled by
   the 10-minute unfilled timeout.  Backtests assume every signal fills;
   a low live fill rate breaks that assumption.
3. Open-order classification and entry fill latency: distinguish protective
   stop-market orders from genuinely pending entry/exit orders, and report
   median/p90/max fill time so a nominal 100% fill rate cannot hide stale fills.

Stdlib only, so it can be piped to the deployment host without installing
anything:

    ssh <host> python3 - --db /data/freqtrade/user_data/tradesv3-v5.sqlite \
        --log /data/freqtrade/user_data/logs/freqtrade.log.1 \
        --log /data/freqtrade/user_data/logs/freqtrade.log \
        --since 2026-07-17T03:49:02Z \
        < freqtrade/scripts/slippage_audit.py

The log is required for a truthful fill-rate denominator because Freqtrade
deletes fully cancelled, zero-fill entry trades and their orders from SQLite.
"""

from __future__ import annotations

import argparse
import re
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from statistics import mean

STOP_REASONS = ("stop_loss", "stoploss_on_exchange", "trailing_stop_loss")
ENTRY_DELETE_RE = re.compile(
    r"^(?P<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d+ .* "
    r"(?P<side>Buy|Sell) order fully cancelled\. Removing Trade\("
    r"id=(?P<trade_id>\d+), pair=(?P<pair>[^,]+), amount=0, "
    r"is_short=(?P<is_short>True|False),.*open_rate=(?P<open_rate>[^,]+),"
)


def normalize_since(value: str | None) -> str | None:
    if value is None:
        return None
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is not None:
        parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None)
    return parsed.isoformat(sep=" ", timespec="microseconds")


def parse_deleted_entry_timeouts(
    log_paths: str | list[str] | None, since: str | None
) -> list[dict]:
    """Recover fully cancelled entries that Freqtrade deletes from SQLite."""
    if log_paths is None:
        return []
    if isinstance(log_paths, str):
        log_paths = [log_paths]
    cutoff = datetime.fromisoformat(since) if since else None
    rows = []
    seen: set[tuple[datetime, str, str, float]] = set()
    for log_path in log_paths:
        with Path(log_path).open(encoding="utf-8", errors="replace") as handle:
            for line in handle:
                match = ENTRY_DELETE_RE.search(line)
                if match is None:
                    continue
                event_time = datetime.fromisoformat(match.group("time"))
                if cutoff is not None and event_time < cutoff:
                    continue
                pair = match.group("pair")
                side = "short" if match.group("is_short") == "True" else "long"
                open_rate = float(match.group("open_rate"))
                key = (event_time, pair, side, open_rate)
                if key in seen:
                    continue
                seen.add(key)
                rows.append(
                    {
                        "time": event_time,
                        "pair": pair,
                        "side": side,
                        "open_rate": open_rate,
                    }
                )
    return sorted(rows, key=lambda row: (row["time"], row["pair"], row["side"]))


def fetch_rows(con: sqlite3.Connection, query: str, params: tuple = ()) -> list[sqlite3.Row]:
    con.row_factory = sqlite3.Row
    return list(con.execute(query, params))


def audit_stop_slippage(con: sqlite3.Connection, since: str | None = None) -> None:
    rows = fetch_rows(
        con,
        """
        SELECT id, pair, is_short, leverage, exit_reason,
               stop_loss, close_rate, close_profit, close_date
        FROM trades
        WHERE is_open = 0 AND exit_reason IN (?, ?, ?)
              AND stop_loss > 0 AND close_rate > 0
              AND (? IS NULL OR close_date >= ?)
        ORDER BY close_date
        """,
        (*STOP_REASONS, since, since),
    )
    if not rows:
        print("stop slippage: no closed stop exits yet")
        return

    print(f"stop slippage ({len(rows)} closed stop exits, + = adverse cost):")
    per_reason: dict[str, list[float]] = {}
    for r in rows:
        if r["is_short"]:
            slip = (r["close_rate"] - r["stop_loss"]) / r["stop_loss"]
        else:
            slip = (r["stop_loss"] - r["close_rate"]) / r["stop_loss"]
        per_reason.setdefault(r["exit_reason"], []).append(slip)
        print(
            f"  #{r['id']:<3} {r['pair']:<18} {r['exit_reason']:<22}"
            f" stop={r['stop_loss']:<12.6g} fill={r['close_rate']:<12.6g}"
            f" slip={slip * 100:+.3f}%"
            f" leveraged={slip * (r['leverage'] or 1) * 100:+.3f}%"
            f"  close_profit={r['close_profit'] * 100:+.2f}%"
        )
    for reason, slips in sorted(per_reason.items()):
        print(
            f"  -> {reason}: n={len(slips)}"
            f" mean={mean(slips) * 100:+.3f}% worst={max(slips) * 100:+.3f}%"
        )


def percentile(values: list[float], quantile: float) -> float:
    """Linear-interpolated percentile without third-party dependencies."""
    ordered = sorted(values)
    if not ordered:
        raise ValueError("percentile requires at least one value")
    position = (len(ordered) - 1) * quantile
    lower = int(position)
    upper = min(lower + 1, len(ordered) - 1)
    weight = position - lower
    return ordered[lower] * (1 - weight) + ordered[upper] * weight


def audit_entry_fill_rate(
    con: sqlite3.Connection,
    since: str | None = None,
    deleted_timeouts: list[dict] | None = None,
) -> None:
    rows = fetch_rows(
        con,
        """
        SELECT o.status, o.order_type, o.ft_order_side, o.filled, o.amount,
               o.ft_pair, o.order_date, o.order_filled_date, t.is_short
        FROM orders o JOIN trades t ON t.id = o.ft_trade_id
        WHERE o.order_type = 'limit'
              AND (? IS NULL OR o.order_date >= ?)
        """,
        (since, since),
    )
    entries = [
        r for r in rows
        if r["ft_order_side"] == ("sell" if r["is_short"] else "buy")
    ]
    deleted_timeouts = deleted_timeouts or []
    if not entries and not deleted_timeouts:
        print("entry fill rate: no limit entry orders yet")
        return

    filled = sum(1 for r in entries if r["status"] == "closed")
    partial = sum(
        1 for r in entries
        if r["status"] not in ("closed",) and (r["filled"] or 0) > 0
    )
    unfilled_in_db = len(entries) - filled - partial
    unfilled = unfilled_in_db + len(deleted_timeouts)
    submitted = len(entries) + len(deleted_timeouts)
    print(
        f"entry fill rate: {filled}/{submitted} filled"
        f" ({filled / submitted * 100:.1f}%),"
        f" {partial} partial, {unfilled} unfilled/cancelled"
    )
    if deleted_timeouts:
        print(
            "  deleted timeout entries recovered from log: "
            + ", ".join(
                f"{row['time'].isoformat()} {row['pair']} {row['side']}@{row['open_rate']}"
                for row in deleted_timeouts
            )
        )
    latencies: list[tuple[str, float]] = []
    for row in entries:
        if row["status"] != "closed" or not row["order_filled_date"]:
            continue
        ordered_at = datetime.fromisoformat(row["order_date"])
        filled_at = datetime.fromisoformat(row["order_filled_date"])
        latencies.append((row["ft_pair"], (filled_at - ordered_at).total_seconds()))
    if latencies:
        seconds = [latency for _, latency in latencies]
        print(
            "entry fill latency:"
            f" median={percentile(seconds, 0.5):.1f}s"
            f" p90={percentile(seconds, 0.9):.1f}s"
            f" max={max(seconds):.1f}s"
        )
        slow = [(pair, latency) for pair, latency in latencies if latency >= 60]
        if slow:
            print(
                "  fills >=60s: "
                + ", ".join(f"{pair}={latency:.1f}s" for pair, latency in slow)
            )


def audit_open_orders(con: sqlite3.Connection) -> None:
    rows = fetch_rows(
        con,
        """
        SELECT o.id, o.ft_pair, o.ft_order_side, o.order_type, o.status,
               o.amount, o.filled, o.stop_price, o.order_date, o.order_update_date,
               t.is_short
        FROM orders o JOIN trades t ON t.id = o.ft_trade_id
        WHERE o.ft_is_open = 1
        ORDER BY o.order_date
        """,
    )
    if not rows:
        print("open orders: none")
        return
    print(f"open orders: {len(rows)}")
    for row in rows:
        role = "protective-stop" if row["ft_order_side"] == "stoploss" else "entry/exit"
        print(
            f"  #{row['id']} {row['ft_pair']} {role} {row['order_type']}"
            f" status={row['status']} amount={row['amount']} filled={row['filled']}"
            f" stop={row['stop_price']} updated={row['order_update_date']}"
        )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--db", required=True, help="Path to tradesv3-*.sqlite")
    parser.add_argument(
        "--log",
        action="append",
        help="Freqtrade log used to recover deleted entry timeouts; repeat for rotations",
    )
    parser.add_argument("--since", help="Only include orders/exits at or after this ISO timestamp")
    args = parser.parse_args()

    since = normalize_since(args.since)
    con = sqlite3.connect(f"file:{args.db}?mode=ro", uri=True)
    n_closed = con.execute(
        "SELECT count(*) FROM trades WHERE is_open = 0 AND (? IS NULL OR close_date >= ?)",
        (since, since),
    ).fetchone()[0]
    print(f"db={args.db} since={since or '-'} closed_trades={n_closed}")
    audit_open_orders(con)
    audit_stop_slippage(con, since)
    deleted_timeouts = parse_deleted_entry_timeouts(args.log, since)
    audit_entry_fill_rate(con, since, deleted_timeouts)
    if n_closed < 30:
        print(f"NOTE: only {n_closed} closed trades — treat aggregates as provisional (<30).")
    return 0


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