clboss/contrib/plot-fees

#!/usr/bin/env python3
import argparse
import csv
import json
import math
import os
import re
import subprocess
import shutil
import sys
from datetime import datetime, timedelta, timezone

from clboss.alias_cache import is_nodeid, load_cache, lookup_alias, lookup_nodeid_by_alias
from feemon_data import load_merged_peer_records, records_to_rows

PLOT_LINEWIDTH = 1.5
SCID_RE = re.compile(r"^\d+x\d+x\d+$")
FEE_SYMLOG_LINTHRESH = 1.0
FEE_SYMLOG_LINSCALE = 0.3
EARNINGS_SYMLOG_LINTHRESH = 1.0
EARNINGS_SYMLOG_LINSCALE = 0.3

try:
    import numpy as np
except ImportError:
    np = None


def local_tz():
    return timezone.utc


def normalize_dt(dt):
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def parse_ts_iso(ts):
    if ts.endswith("Z"):
        ts = ts[:-1] + "+00:00"
    return normalize_dt(datetime.fromisoformat(ts))


def parse_ts_epoch(ts):
    if isinstance(ts, (int, float)):
        return datetime.fromtimestamp(ts, timezone.utc)
    try:
        return datetime.fromtimestamp(float(ts), timezone.utc)
    except (TypeError, ValueError):
        return parse_ts_iso(ts)


def parse_ts(ts):
    return parse_ts_epoch(ts)


def make_engineering_formatter():
    try:
        from matplotlib.ticker import ScalarFormatter
    except Exception:
        return None

    class EngineeringScalarFormatter(ScalarFormatter):
        def _set_order_of_magnitude(self, *args, **kwargs):
            super()._set_order_of_magnitude(*args, **kwargs)
            order = self.orderOfMagnitude
            if order != 0:
                self.orderOfMagnitude = int(math.floor(order / 3.0) * 3)

    formatter = EngineeringScalarFormatter(useMathText=True)
    formatter.set_scientific(True)
    formatter.set_powerlimits((0, 0))
    return formatter


def safe_prefix(value):
    sanitized = []
    for ch in value:
        if ch.isalnum() or ch in "._-":
            sanitized.append(ch)
        else:
            sanitized.append("_")
    return "".join(sanitized).strip("_") or "peer"


def is_scid(value):
    return bool(SCID_RE.match(value))


def parse_time_arg(value):
    if value is None:
        return None
    # Ideally, we'd accept the same time arguments as journalctl.
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    if re.match(r"^[0-9]{4}-[0-9]{2}$", value):
        return normalize_dt(datetime.fromisoformat(f"{value}-01T00:00:00"))
    if re.match(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}$", value):
        return normalize_dt(datetime.fromisoformat(f"{value}T00:00:00"))
    rel = re.match(r"^([+-]?)(\d+)([smhdw])$", value)
    if rel:
        sign, num_s, unit = rel.groups()
        num = int(num_s)
        delta = {
            "s": timedelta(seconds=num),
            "m": timedelta(minutes=num),
            "h": timedelta(hours=num),
            "d": timedelta(days=num),
            "w": timedelta(weeks=num),
        }[unit]
        if sign == "-":
            delta = -delta
        return normalize_dt(datetime.now(timezone.utc) + delta)
    if value.count(":") in (1, 2) and "T" not in value and "-" not in value:
        today = datetime.now(timezone.utc).date().isoformat()
        return normalize_dt(datetime.fromisoformat(f"{today}T{value}"))
    return normalize_dt(datetime.fromisoformat(value))


def price_level_to_mult(level):
    if np is None:
        if isinstance(level, (int, float)):
            if level < 0:
                return math.pow(0.8, -level)
            if level > 0:
                return math.pow(1.2, level)
            return 1.0
        return [price_level_to_mult(v) for v in level]

    arr = np.asarray(level, dtype=float)
    mult = np.where(
        arr < 0,
        np.power(0.8, -arr),
        np.where(arr > 0, np.power(1.2, arr), 1.0),
    )
    if np.isscalar(level):
        return float(mult)
    return mult


def price_mult_to_level(mult):
    if np is None:
        if isinstance(mult, (int, float)):
            if mult < 1:
                return -math.log(mult) / math.log(0.8)
            if mult > 1:
                return math.log(mult) / math.log(1.2)
            return 0.0
        return [price_mult_to_level(v) for v in mult]

    arr = np.asarray(mult, dtype=float)
    level = np.zeros_like(arr, dtype=float)
    lt_one = (arr > 0) & (arr < 1)
    gt_one = arr > 1
    if np.any(lt_one):
        level[lt_one] = -np.log(arr[lt_one]) / np.log(0.8)
    if np.any(gt_one):
        level[gt_one] = np.log(arr[gt_one]) / np.log(1.2)
    if np.isscalar(mult):
        return float(level)
    return level


def format_mult_tick(value):
    if value == 0:
        return "0"
    if value < 1:
        nice_values = [
            (0.5, "0.5"),
            (0.2, "0.2"),
            (0.1, "0.10"),
            (0.05, "0.05"),
            (0.02, "0.02"),
            (0.01, "0.01"),
        ]
        for target, label in nice_values:
            if abs(value - target) / target <= 0.08:
                return label
    abs_val = abs(value)
    for decimals in (0, 1, 2):
        rounded = round(value, decimals)
        if rounded != 0 and abs(value - rounded) / abs_val < 0.01:
            if decimals == 0:
                return str(int(round(rounded)))
            text = f"{rounded:.{decimals}f}"
            return text.rstrip("0").rstrip(".")
    return f"{value:.3g}"


def nice_mult_ticks(vmin, vmax):
    if vmin <= 0 or vmax <= 0:
        return []
    if vmin > vmax:
        vmin, vmax = vmax, vmin
    exp_min = int(math.floor(math.log10(vmin)))
    exp_max = int(math.ceil(math.log10(vmax)))
    mantissas = [1, 2, 5]
    ticks = []
    for exp in range(exp_min, exp_max + 1):
        base = 10 ** exp
        for m in mantissas:
            value = m * base
            if vmin <= value <= vmax:
                ticks.append(value)
    if not ticks:
        if vmin == vmax:
            return [vmin]
        return [vmin, vmax]
    return ticks


def pad_top(ax, frac=0.2):
    ymin, ymax = ax.get_ylim()
    data_max = ax.dataLim.ymax
    if data_max <= 0:
        return
    target_top = data_max * (1.0 + frac)
    ax.set_ylim(bottom=ymin, top=target_top)


def set_log_ylim(ax, values, pad_frac=0.2, min_floor=0.9, min_top=10):
    if not values:
        return
    max_val = max(values)
    if max_val < min_floor:
        max_val = min_floor
    top = max(max_val * (1.0 + pad_frac), min_top)
    bottom = min_floor
    ax.set_ylim(bottom=bottom, top=top)


def set_log_ylim_from_ax(ax, pad_frac=0.2, min_floor=0.9, min_top=10):
    max_val = ax.dataLim.ymax
    if max_val <= 0:
        return
    top = max(max_val * (1.0 + pad_frac), min_top)
    bottom = min_floor
    ax.set_ylim(bottom=bottom, top=top)


def set_log_decade_ticks(ax):
    ymin, ymax = ax.get_ylim()
    if ymin <= 0 or ymax <= 0:
        return
    exp_min = int(math.floor(math.log10(ymin)))
    exp_max = int(math.ceil(math.log10(ymax)))
    ticks = [10 ** e for e in range(exp_min, exp_max + 1) if 10 ** e >= 1]
    labels = [str(int(t)) for t in ticks]
    ax.set_yticks(ticks)
    ax.set_yticklabels(labels)


def set_symlog_ylim(
    ax,
    values,
    clamp_zero=False,
    pad_frac=0.1,
    min_pad=1.0,
    min_bottom=None,
    max_bottom=None,
    min_top=None,
):
    if not values:
        return
    vmin = min(values)
    vmax = max(values)
    if vmin == vmax:
        pad = max(abs(vmin) * pad_frac, min_pad)
    else:
        pad = (vmax - vmin) * pad_frac
    bottom = vmin - pad
    top = vmax + pad
    if clamp_zero and bottom < 0:
        bottom = 0
    if min_bottom is not None and bottom > min_bottom:
        bottom = min_bottom
    if max_bottom is not None and bottom < max_bottom:
        bottom = max_bottom
    if min_top is not None and top < min_top:
        top = min_top
    ax.set_ylim(bottom=bottom, top=top)


def set_symlog_ticks(ax, linthresh, symmetric=True):
    ymin, ymax = ax.get_ylim()
    max_abs = max(abs(ymin), abs(ymax))
    if max_abs <= 0:
        ax.set_yticks([0])
        ax.set_yticklabels(["0"])
        return
    max_exp = int(math.floor(math.log10(max_abs)))
    ticks = [0.0]
    for exp in range(0, max_exp + 1):
        tick = 10 ** exp
        if tick < linthresh or tick > max_abs:
            continue
        if symmetric:
            ticks.extend([-tick, tick])
        else:
            ticks.append(tick)
    ticks = sorted(set(ticks))
    labels = []
    for tick in ticks:
        if tick == 0:
            labels.append("0")
        else:
            labels.append(str(int(tick)))
    ax.set_yticks(ticks)
    ax.set_yticklabels(labels)


def add_common_args(parser):
    parser.add_argument(
        "--db",
        default=None,
        help="Path to legacy sqlite database (optional; default: API only).",
    )
    parser.add_argument(
        "--png",
        nargs="?",
        const="__DEFAULT__",
        default=None,
        help="Output image path.",
    )
    parser.add_argument(
        "--csv",
        nargs="?",
        const="__DEFAULT__",
        default=None,
        help="Output CSV path.",
    )
    parser.add_argument(
        "--show",
        action="store_true",
        help="Show the plot interactively.",
    )
    parser.add_argument(
        "--title",
        nargs="?",
        const="",
        default=None,
        help="Plot title (defaults to peer label; pass empty to omit).",
    )
    parser.add_argument(
        "--since",
        dest="from_ts",
        default=None,
        help="Start time (ISO-8601 or HH:MM[:SS]).",
    )
    parser.add_argument(
        "--before",
        dest="to_ts",
        default=None,
        help="End time (ISO-8601 or HH:MM[:SS]).",
    )


def add_lightning_args(parser):
    parser.add_argument("--mainnet", action="store_true", help="Run on mainnet")
    parser.add_argument("--testnet", action="store_true", help="Run on testnet")
    parser.add_argument("--signet", action="store_true", help="Run on signet")
    parser.add_argument("--regtest", action="store_true", help="Run on regtest")
    parser.add_argument("--network", help="Set the network explicitly")
    parser.add_argument("--lightning-dir", help="lightning data location")


def resolve_network_option(args):
    if args.network:
        return f"--network={args.network}"
    if args.testnet:
        return "--network=testnet"
    if args.signet:
        return "--network=signet"
    if args.regtest:
        return "--network=regtest"
    return "--network=bitcoin"


def resolve_lightning_dir(args):
    if args.lightning_dir:
        if not os.path.isdir(args.lightning_dir):
            raise ValueError(f'"{args.lightning_dir}" is not a valid directory')
        return args.lightning_dir
    return None


def run_lightning_cli_command(lightning_dir, network_option, subcommand, *args):
    try:
        command = ["lightning-cli", network_option, subcommand, *args]
        if lightning_dir:
            command = command[:2] \
                + [f"--lightning-dir={lightning_dir}"] \
                + command[2:]
        result = subprocess.run(command, capture_output=True, text=True, check=True)
        return json.loads(result.stdout)
    except subprocess.CalledProcessError as e:
        print(f"Command '{command}' failed with error: {e}", file=sys.stderr)
    except FileNotFoundError:
        print("lightning-cli not found in PATH.", file=sys.stderr)
    except json.JSONDecodeError as e:
        print(f"Failed to parse JSON from command '{command}': {e}", file=sys.stderr)
    return None


def lookup_nodeid_by_scid(run_lightning_cli_command, lightning_dir, network_option, scid):
    listpeerchannels_data = run_lightning_cli_command(
        lightning_dir, network_option, "listpeerchannels"
    )
    if not listpeerchannels_data:
        return None
    channels = listpeerchannels_data.get("channels", [])
    for channel in channels:
        if channel.get("short_channel_id") == scid:
            return channel.get("peer_id")
    return None


def resolve_peer(args):
    peer_input = args.peer
    if is_nodeid(peer_input):
        return peer_input, peer_input

    cli_available = shutil.which("lightning-cli") is not None
    network_option = args.network_option
    lightning_dir = args.resolved_lightning_dir

    if is_scid(peer_input):
        if not cli_available:
            raise SystemExit(
                "Error: lightning-cli not found. Provide a nodeid for SCID inputs."
            )
        peer_id = lookup_nodeid_by_scid(
            run_lightning_cli_command, lightning_dir, network_option, peer_input
        )
        if not peer_id:
            raise SystemExit(f"Error: SCID '{peer_input}' not found.")
        alias = lookup_alias(
            run_lightning_cli_command, lightning_dir, network_option, peer_id
        )
        if alias and alias != peer_id:
            return peer_id, f"{alias} ({peer_input})"
        return peer_id, peer_input

    alias = peer_input
    if cli_available:
        peer_id = lookup_nodeid_by_alias(
            run_lightning_cli_command, lightning_dir, network_option, alias
        )
    else:
        cache = load_cache()
        peer_id = None
        for node_id, cached_alias in cache.items():
            if cached_alias == alias:
                peer_id = node_id
                break
    if not peer_id:
        if not cli_available:
            raise SystemExit(
                "Error: lightning-cli not found and alias not in cache. "
                "Provide a nodeid or ensure lightning-cli is in PATH."
            )
        raise SystemExit(f"Error: Alias '{alias}' not found.")
    return peer_id, alias


def filter_by_time(rows, from_ts, to_ts):
    if from_ts is None and to_ts is None:
        return rows
    filtered = []
    for row in rows:
        ts = parse_ts(row[0])
        if from_ts and ts < from_ts:
            continue
        if to_ts and ts > to_ts:
            continue
        filtered.append(row)
    return filtered


def fetch_earnings_rows(args):
    if shutil.which("lightning-cli") is None:
        print("warning: lightning-cli not found; skipping earnings plot", file=sys.stderr)
        return []

    network_option = args.network_option
    lightning_dir = args.resolved_lightning_dir
    data = run_lightning_cli_command(
        lightning_dir, network_option, "clboss-earnings-history", args.peer
    )
    if not data or "history" not in data:
        print("warning: earnings data unavailable; skipping earnings plot", file=sys.stderr)
        return []

    rows = []
    for entry in data["history"]:
        bucket_time = entry.get("bucket_time")
        if not bucket_time:
            continue
        in_earnings = entry.get("in_earnings", 0)
        out_earnings = entry.get("out_earnings", 0)
        rows.append((bucket_time, in_earnings, out_earnings))

    return filter_by_time(rows, args.from_ts, args.to_ts)


def get_out_path(args, suffix):
    if args.png is None:
        return None
    if args.png == "__DEFAULT__":
        prefix = safe_prefix(args.peer_input)[:32]
        return f"{prefix}-{suffix}.png"
    return args.png


def get_csv_path(args, suffix):
    if args.csv is None:
        return None
    if args.csv == "__DEFAULT__":
        prefix = safe_prefix(args.peer_input)[:32]
        return f"{prefix}-{suffix}.csv"
    return args.csv


def write_csv(args, rows, suffix):
    path = get_csv_path(args, suffix)
    if not path:
        return
    headers = [
        "ts",
        "peer",
        "set_base",
        "set_ppm",
        "baseline_base",
        "baseline_ppm",
        "size_mult",
        "size_total_peers",
        "size_less_peers",
        "balance_mult",
        "balance_our_msat",
        "balance_total_msat",
        "price_level",
        "price_mult",
    ]
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(headers)
        writer.writerows(rows)


CSV_FIELDS = [
    "node_id",
    "set_base",
    "set_ppm",
    "baseline_base",
    "baseline_ppm",
    "size_mult",
    "size_total_peers",
    "size_less_peers",
    "balance_mult",
    "balance_our_msat",
    "balance_total_msat",
    "price_level",
    "price_mult",
]
PRICE_LEVEL_FIELDS = ["price_level", "price_mult", "price_center"]
BASE_PPM_FIELDS = ["set_base", "set_ppm"]
BASELINE_FIELDS = ["baseline_base", "baseline_ppm"]
SIZE_MULT_FIELDS = ["size_total_peers", "size_less_peers", "size_mult"]
BALANCE_MULT_FIELDS = ["balance_our_msat", "balance_total_msat", "balance_mult"]
WARNED_FEEMON_MESSAGES = set()


def warn_feemon_data(message):
    if message in WARNED_FEEMON_MESSAGES:
        return
    WARNED_FEEMON_MESSAGES.add(message)
    print(f"warning: {message}", file=sys.stderr)


def get_fee_records(args):
    if not hasattr(args, "_fee_records"):
        args._fee_records = load_merged_peer_records(
            args.db,
            args.peer,
            since_dt=args.from_ts,
            before_dt=args.to_ts,
            lightning_dir=args.resolved_lightning_dir,
            network_option=args.network_option,
            warn=warn_feemon_data,
        )
    return args._fee_records


def fetch_rows(args, fields):
    return records_to_rows(get_fee_records(args), fields)


def fetch_csv_rows(args):
    return fetch_rows(args, CSV_FIELDS)


def make_single_axes(plt):
    fig, ax = plt.subplots(figsize=(12, 4))
    return fig, ax, (ax,)


def make_twin_axes(plt):
    fig, ax_left = plt.subplots(figsize=(12, 4))
    ax_right = ax_left.twinx()
    return fig, ax_left, (ax_left, ax_right)


def plot_single_view(
    args, suffix, row_fields, axes_factory, plotter, use_plt_tight_layout=False
):
    try:
        import matplotlib.pyplot as plt
    except ImportError:
        raise SystemExit("matplotlib is required to output graphs.")
    else:
        import matplotlib as mpl

        mpl.rcParams["timezone"] = "UTC"

    rows = fetch_rows(args, row_fields)
    csv_rows = fetch_csv_rows(args)

    fig, primary_ax, plot_axes = axes_factory(plt)
    plotter(*plot_axes, rows, args.peer_label)
    primary_ax.set_xlabel("time")
    primary_ax.set_title(args.title)

    if use_plt_tight_layout:
        plt.tight_layout()
    else:
        fig.tight_layout()
    out_path = get_out_path(args, suffix)
    if out_path:
        plt.savefig(out_path, dpi=150)

    if args.show:
        plt.show()
    write_csv(args, csv_rows, suffix)


def plot_price_level_on(ax_level, rows, peer):
    rows = [r for r in rows if r[1] is not None]
    ts_levels = []
    levels = []
    centers = []
    has_center = False
    for t, level, _, center in rows:
        ts = parse_ts(t)
        ts_levels.append(ts)
        levels.append(level)
        if center is None:
            centers.append(float("nan"))
        else:
            has_center = True
            centers.append(float(center))

    if not ts_levels:
        raise SystemExit("no theory_level data for peer in time range")

    level_line, = ax_level.plot(
        ts_levels,
        levels,
        linewidth=PLOT_LINEWIDTH,
        label="theory_level",
        color="orange",
    )
    center_line = None
    if has_center:
        center_line, = ax_level.plot(
            ts_levels,
            centers,
            linewidth=PLOT_LINEWIDTH,
            label="theory_center",
            color="red",
            alpha=0.85,
        )

    ax_level.set_ylabel("theory_level")
    ax_mult = ax_level.secondary_yaxis(
        "right",
        functions=(price_level_to_mult, price_mult_to_level),
    )
    ax_mult.set_ylabel("theory_mult")
    if center_line is None:
        ax_level.legend([level_line], ["theory_level"])
    else:
        ax_level.legend(
            [level_line, center_line],
            ["theory_level", "theory_center"],
        )
    pad_top(ax_level)
    scale_levels = levels
    if has_center:
        scale_levels = levels + [v for v in centers if not math.isnan(v)]
    mult_min = price_level_to_mult(min(scale_levels))
    mult_max = price_level_to_mult(max(scale_levels))
    mult_ticks = nice_mult_ticks(mult_min, mult_max)
    if mult_ticks:
        ax_mult.set_yticks(mult_ticks)
        ax_mult.set_yticklabels([format_mult_tick(v) for v in mult_ticks])
    return ax_mult


def plot_base_ppm_on(ax_base, ax_ppm, rows, peer):

    rows = [r for r in rows if r[1] is not None and r[2] is not None]
    ts = []
    bases = []
    ppms = []
    for t, base, ppm in rows:
        ts.append(parse_ts(t))
        bases.append(base)
        ppms.append(ppm)

    if not ts:
        raise SystemExit("no est_base/est_ppm data for peer in time range")

    base_line, = ax_base.plot(ts, bases, linewidth=PLOT_LINEWIDTH, label="base")
    ppm_line, = ax_ppm.plot(ts, ppms, linewidth=PLOT_LINEWIDTH, label="ppm", color="orange")

    ax_base.set_ylabel("advertised_base")
    ax_ppm.set_ylabel("advertised_ppm")
    ax_base.set_yscale(
        "symlog",
        linthresh=FEE_SYMLOG_LINTHRESH,
        linscale=FEE_SYMLOG_LINSCALE,
    )
    ax_ppm.set_yscale(
        "symlog",
        linthresh=FEE_SYMLOG_LINTHRESH,
        linscale=FEE_SYMLOG_LINSCALE,
    )

    base_values = [0.0] + bases
    ppm_values = [0.0] + ppms
    ppm_scale_values = [max(0.0, v) for v in ppm_values]
    base_min = min(base_values)
    ppm_min = min(ppm_scale_values)
    base_max_bottom = None if base_min < 0 else -0.1
    set_symlog_ylim(
        ax_base,
        base_values,
        min_pad=0.1,
        min_bottom=-0.1,
        max_bottom=base_max_bottom,
        min_top=10,
    )
    set_symlog_ylim(
        ax_ppm,
        ppm_scale_values,
        min_pad=0.1,
        min_bottom=-0.1,
        max_bottom=-0.1,
        min_top=10,
    )
    set_symlog_ticks(ax_base, FEE_SYMLOG_LINTHRESH, symmetric=base_min < 0)
    set_symlog_ticks(ax_ppm, FEE_SYMLOG_LINTHRESH, symmetric=ppm_min < 0)

    ax_base.legend([base_line, ppm_line], ["advertised_base", "advertised_ppm"])


def plot_baseline_on(ax_base, ax_ppm, rows, peer):

    rows = [r for r in rows if r[1] is not None and r[2] is not None]
    ts = []
    bases = []
    ppms = []
    for t, base, ppm in rows:
        ts.append(parse_ts(t))
        bases.append(base)
        ppms.append(ppm)

    if not ts:
        raise SystemExit("no baseline data for peer in time range")

    base_line, = ax_base.plot(ts, bases, linewidth=PLOT_LINEWIDTH, label="base")
    ppm_line, = ax_ppm.plot(ts, ppms, linewidth=PLOT_LINEWIDTH, label="ppm", color="orange")

    ax_base.set_ylabel("baseline_base")
    ax_ppm.set_ylabel("baseline_ppm")

    ax_base.set_yscale(
        "symlog",
        linthresh=FEE_SYMLOG_LINTHRESH,
        linscale=FEE_SYMLOG_LINSCALE,
    )
    ax_ppm.set_yscale(
        "symlog",
        linthresh=FEE_SYMLOG_LINTHRESH,
        linscale=FEE_SYMLOG_LINSCALE,
    )

    base_values = [0.0] + bases
    ppm_values = [0.0] + ppms
    ppm_scale_values = [max(0.0, v) for v in ppm_values]
    base_min = min(base_values)
    ppm_min = min(ppm_scale_values)
    base_max_bottom = None if base_min < 0 else -0.1
    set_symlog_ylim(
        ax_base,
        base_values,
        min_pad=0.1,
        min_bottom=-0.1,
        max_bottom=base_max_bottom,
        min_top=10,
    )
    set_symlog_ylim(
        ax_ppm,
        ppm_scale_values,
        min_pad=0.1,
        min_bottom=-0.1,
        max_bottom=-0.1,
        min_top=10,
    )
    set_symlog_ticks(ax_base, FEE_SYMLOG_LINTHRESH, symmetric=base_min < 0)
    set_symlog_ticks(ax_ppm, FEE_SYMLOG_LINTHRESH, symmetric=ppm_min < 0)

    ax_base.legend([base_line, ppm_line], ["baseline_base", "baseline_ppm"])


def plot_size_mult_on(ax_left, ax_right, rows, peer):
    rows = [
        r for r in rows
        if r[1] is not None and r[2] is not None and r[3] is not None
    ]
    ts = []
    totals = []
    lesses = []
    mults = []
    for t, total, less, mult in rows:
        ts.append(parse_ts(t))
        totals.append(total)
        lesses.append(less)
        mults.append(mult)

    if not ts:
        raise SystemExit("no size_total_peers data for peer in time range")

    total_line, = ax_left.plot(
        ts, totals, linewidth=PLOT_LINEWIDTH, label="total_peers", color="green"
    )
    less_line, = ax_left.plot(
        ts, lesses, linewidth=PLOT_LINEWIDTH, label="less_peers"
    )
    mult_line, = ax_right.plot(
        ts, mults, linewidth=PLOT_LINEWIDTH, label="size_mult", color="orange"
    )
    ax_left.set_ylabel("peer_count")
    ax_left.set_ylim(bottom=0)
    pad_top(ax_left)
    ax_left.legend(
        [total_line, less_line, mult_line],
        ["total_peers", "less_peers", "size_mult"],
    )
    ax_right.set_ylabel("size_mult")
    ax_right.set_yscale("log")
    ax_right.set_ylim(0.5, 16)
    ax_right.set_yticks(
        [0.5, 1, 2, 4, 8, 16],
        ["1/2", "1", "2", "4", "8", "16"],
    )


def plot_balance_mult_on(ax_left, ax_right, rows, peer):
    rows = [
        r for r in rows
        if r[1] is not None and r[2] is not None and r[3] is not None
    ]
    ts = []
    our_sats = []
    total_sats = []
    mults = []
    for t, our_val, total_val, mult in rows:
        ts.append(parse_ts(t))
        our_sats.append(our_val / 1000.0)
        total_sats.append(total_val / 1000.0)
        mults.append(mult)

    if not ts:
        raise SystemExit("no balance data for peer in time range")

    total_line, = ax_left.plot(
        ts, total_sats, linewidth=PLOT_LINEWIDTH, label="total_balance", color="green"
    )
    peer_line, = ax_left.plot(
        ts, our_sats, linewidth=PLOT_LINEWIDTH, label="our_balance"
    )
    mult_line, = ax_right.plot(
        ts, mults, linewidth=PLOT_LINEWIDTH, color="orange", label="balance_mult"
    )
    ax_left.set_ylabel("balance")
    max_total = max(total_sats) if total_sats else 0
    if max_total > 0:
        ax_left.set_ylim(bottom=0 - 0.05, top=max_total * 1.05)
    else:
        ax_left.set_ylim(bottom=0 - 0.05)
    balance_formatter = make_engineering_formatter()
    if balance_formatter:
        ax_left.yaxis.set_major_formatter(balance_formatter)
    ax_left.legend(
        [total_line, peer_line, mult_line],
        ["total_balance", "our_balance", "balance_mult"],
    )
    ax_right.set_ylabel("balance_mult")
    ax_right.set_yscale("log")
    ax_right.set_ylim(1 / 6 * 0.8, 6 * 1.2)
    ax_right.set_yticks(
        [1 / 6, 1 / 3, 1, 3, 6],
        ["1/6", "1/3", "1", "3", "6"],
    )


def plot_earnings_split(ax_out, ax_in, rows, peer):
    ts = []
    incoming = []
    outgoing = []
    for t, in_earnings, out_earnings in rows:
        ts.append(parse_ts(t))
        incoming.append(in_earnings / 1000.0)
        outgoing.append(out_earnings / 1000.0)

    if not ts:
        for ax, label in ((ax_out, "outgoing"), (ax_in, "incoming")):
            ax.set_ylabel(f"earnings (sat/day) {label}")
            ax.set_yscale(
                "symlog",
                linthresh=EARNINGS_SYMLOG_LINTHRESH,
                linscale=EARNINGS_SYMLOG_LINSCALE,
            )
            set_symlog_ylim(
                ax,
                [0.0],
                clamp_zero=True,
                min_bottom=0,
                min_top=EARNINGS_SYMLOG_LINTHRESH,
            )
            set_symlog_ticks(ax, EARNINGS_SYMLOG_LINTHRESH, symmetric=False)
        ax_out.set_title("earnings (no data)")
        return

    # Earnings buckets are daily; draw near-full-day bars starting at the bucket
    # boundary to preserve UTC alignment while still visually separating days.
    bar_width = timedelta(days=0.8)
    out_bars = ax_out.bar(
        ts,
        outgoing,
        width=bar_width,
        align="edge",
        color="green",
        label="earnings (sat/day) outgoing",
    )
    in_bars = ax_in.bar(
        ts,
        incoming,
        width=bar_width,
        align="edge",
        color="blue",
        label="earnings (sat/day) incoming",
    )

    for ax, values, label in (
        (ax_out, outgoing, "outgoing"),
        (ax_in, incoming, "incoming"),
    ):
        ax.set_ylabel(f"earnings (sat/day) {label}")
        ax.set_yscale(
            "symlog",
            linthresh=EARNINGS_SYMLOG_LINTHRESH,
            linscale=EARNINGS_SYMLOG_LINSCALE,
        )
        set_symlog_ylim(
            ax,
            [0.0] + values,
            clamp_zero=True,
            min_bottom=0,
            min_top=EARNINGS_SYMLOG_LINTHRESH,
        )
        set_symlog_ticks(ax, EARNINGS_SYMLOG_LINTHRESH, symmetric=False)

    ax_out.legend(handles=[out_bars], labels=["earnings (sat/day) outgoing"])
    ax_in.legend(handles=[in_bars], labels=["earnings (sat/day) incoming"])


def plot_earnings_single_on(ax, rows, direction):
    incoming = direction == "incoming"
    label = "incoming" if incoming else "outgoing"
    color = "blue" if incoming else "green"
    ts = []
    values = []
    for t, in_earnings, out_earnings in rows:
        ts.append(parse_ts(t))
        value = in_earnings if incoming else out_earnings
        values.append(value / 1000.0)

    ax.set_ylabel(f"earnings (sat/day) {label}")
    ax.set_yscale(
        "symlog",
        linthresh=EARNINGS_SYMLOG_LINTHRESH,
        linscale=EARNINGS_SYMLOG_LINSCALE,
    )
    if not ts:
        set_symlog_ylim(
            ax,
            [0.0],
            clamp_zero=True,
            min_bottom=0,
            min_top=EARNINGS_SYMLOG_LINTHRESH,
        )
        set_symlog_ticks(ax, EARNINGS_SYMLOG_LINTHRESH, symmetric=False)
        ax.set_title(f"earnings (no data) {label}")
        return

    bar_width = timedelta(days=0.8)
    bars = ax.bar(
        ts,
        values,
        width=bar_width,
        align="edge",
        color=color,
        label=f"earnings (sat/day) {label}",
    )
    set_symlog_ylim(
        ax,
        [0.0] + values,
        clamp_zero=True,
        min_bottom=0,
        min_top=EARNINGS_SYMLOG_LINTHRESH,
    )
    set_symlog_ticks(ax, EARNINGS_SYMLOG_LINTHRESH, symmetric=False)
    ax.legend(handles=[bars], labels=[f"earnings (sat/day) {label}"])


def plot_price_level(args):
    plot_single_view(
        args,
        "theory",
        PRICE_LEVEL_FIELDS,
        make_single_axes,
        plot_price_level_on,
    )


def plot_base_ppm(args):
    plot_single_view(
        args,
        "baseppm",
        BASE_PPM_FIELDS,
        make_twin_axes,
    plot_base_ppm_on,
    )
    # Keep symlog tick formatting from plot_base_ppm_on.


def plot_baseline(args):
    plot_single_view(
        args,
        "baseline",
        BASELINE_FIELDS,
        make_twin_axes,
    plot_baseline_on,
    )
    # Keep symlog tick formatting from plot_baseline_on.


def plot_size_mult(args):
    plot_single_view(
        args,
        "size",
        SIZE_MULT_FIELDS,
        make_twin_axes,
        plot_size_mult_on,
        use_plt_tight_layout=True,
    )


def plot_balance_mult(args):
    plot_single_view(
        args,
        "balance",
        BALANCE_MULT_FIELDS,
        make_twin_axes,
        plot_balance_mult_on,
        use_plt_tight_layout=True,
    )


def plot_combo(args):
    try:
        import matplotlib.pyplot as plt
    except ImportError:
        raise SystemExit("matplotlib is required to output graphs.")
    else:
        import matplotlib as mpl

        mpl.rcParams["timezone"] = "UTC"

    def extend_times(acc, rows, end_pad=None):
        for row in rows:
            ts_val = row[0]
            if ts_val is None:
                continue
            ts = parse_ts(ts_val)
            acc.append(ts)
            if end_pad:
                acc.append(ts + end_pad)

    size_rows = fetch_rows(args, SIZE_MULT_FIELDS)
    baseline_rows = fetch_rows(args, BASELINE_FIELDS)
    balance_rows = fetch_rows(args, BALANCE_MULT_FIELDS)
    price_rows = fetch_rows(args, PRICE_LEVEL_FIELDS)
    base_rows = fetch_rows(args, BASE_PPM_FIELDS)
    csv_rows = fetch_rows(args, CSV_FIELDS)
    earnings_rows = fetch_earnings_rows(args)
    fig, axes = plt.subplots(7, 1, figsize=(12, 21), sharex=True)
    (
        ax_baseline,
        ax_size,
        ax_balance,
        ax_price,
        ax_base,
        ax_earnings_out,
        ax_earnings_in,
    ) = axes

    ax_baseline_base = ax_baseline
    ax_baseline_ppm = ax_baseline_base.twinx()
    plot_baseline_on(
        ax_baseline_base, ax_baseline_ppm, baseline_rows, args.peer_label
    )

    ax_size_right = ax_size.twinx()
    plot_size_mult_on(ax_size, ax_size_right, size_rows, args.peer_label)
    ax_balance_right = ax_balance.twinx()
    plot_balance_mult_on(ax_balance, ax_balance_right, balance_rows, args.peer_label)

    ax_price_level = ax_price
    ax_price_mult = plot_price_level_on(ax_price_level, price_rows, args.peer_label)

    ax_base_fee = ax_base
    ax_base_ppm = ax_base_fee.twinx()
    plot_base_ppm_on(ax_base_fee, ax_base_ppm, base_rows, args.peer_label)

    plot_earnings_split(ax_earnings_out, ax_earnings_in, earnings_rows, args.peer_label)

    all_times = []
    extend_times(all_times, baseline_rows)
    extend_times(all_times, size_rows)
    extend_times(all_times, balance_rows)
    extend_times(all_times, price_rows)
    extend_times(all_times, base_rows)
    extend_times(all_times, earnings_rows, end_pad=timedelta(days=1))
    if all_times:
        ax_baseline.set_xlim(min(all_times), max(all_times))

    fig.suptitle(args.title)
    fig.tight_layout(rect=[0, 0, 1, 0.96])
    fig.subplots_adjust(hspace=0.15)

    out_path = get_out_path(args, "combo")
    if out_path:
        plt.savefig(out_path, dpi=150)

    if args.show:
        plt.show()
    write_csv(args, csv_rows, "combo")


def plot_earnings_single(args, direction, suffix):
    try:
        import matplotlib.pyplot as plt
    except ImportError:
        raise SystemExit("matplotlib is required to output graphs.")
    else:
        import matplotlib as mpl

        mpl.rcParams["timezone"] = "UTC"

    rows = fetch_earnings_rows(args)
    csv_rows = fetch_csv_rows(args)

    fig, ax = plt.subplots(figsize=(12, 4))
    ax.set_title(args.title)
    plot_earnings_single_on(ax, rows, direction)
    ax.set_xlabel("time")
    fig.tight_layout()

    out_path = get_out_path(args, suffix)
    if out_path:
        plt.savefig(out_path, dpi=150)

    if args.show:
        plt.show()
    write_csv(args, csv_rows, suffix)


def plot_earnings_incoming(args):
    plot_earnings_single(args, "incoming", "earnings-incoming")


def plot_earnings_outgoing(args):
    plot_earnings_single(args, "outgoing", "earnings-outgoing")


def main():
    parser = argparse.ArgumentParser(
        description="Plot fee-related time series from merged API and legacy fee history."
    )
    parser.add_argument(
        "--view",
        required=True,
        choices=[
            "theory",
            "base-ppm",
            "baseline",
            "size",
            "balance",
            "incoming-earnings",
            "outgoing-earnings",
            "combo",
        ],
        help="Plot view to render.",
    )
    parser.add_argument(
        "--peer",
        required=True,
        help="Peer nodeid (hex), alias, or SCID.",
    )
    add_lightning_args(parser)
    add_common_args(parser)

    args = parser.parse_args()
    args.network_option = resolve_network_option(args)
    args.resolved_lightning_dir = resolve_lightning_dir(args)
    args.peer_input = args.peer
    args.from_ts = parse_time_arg(args.from_ts)
    args.to_ts = parse_time_arg(args.to_ts)
    args.peer, args.peer_label = resolve_peer(args)
    args.title = args.peer_label if args.title is None else args.title

    if args.db is None and shutil.which("lightning-cli") is None:
        raise SystemExit(
            "no data source available: provide --db or ensure lightning-cli is in PATH"
        )

    if not args.png and not args.show and not args.csv:
        raise SystemExit("use --png to save, --show to display, or --csv to export")

    view_map = {
        "theory": plot_price_level,
        "base-ppm": plot_base_ppm,
        "baseline": plot_baseline,
        "size": plot_size_mult,
        "balance": plot_balance_mult,
        "incoming-earnings": plot_earnings_incoming,
        "outgoing-earnings": plot_earnings_outgoing,
        "combo": plot_combo,
    }
    view_map[args.view](args)


if __name__ == "__main__":
    main()