#!/usr/bin/env python3 from __future__ import annotations import argparse import json from pathlib import Path from typing import Dict, List, Tuple import pandas as pd def read_csv(path: Path) -> pd.DataFrame: if not path.exists() or path.stat().st_size == 0: return pd.DataFrame() return pd.read_csv(path) def best_rows(df: pd.DataFrame, n: int = 10) -> pd.DataFrame: if df.empty: return df x = df.copy() if "valid_capacity_avg_p95_p99" in x.columns: x = x[x["valid_capacity_avg_p95_p99"].astype(bool)] if "mdd_pct" in x.columns: x = x[x["mdd_pct"].abs() <= 20] if "return_pct" in x.columns: x = x[x["return_pct"] > 0] if x.empty: return df.sort_values("score", ascending=False).head(n) return x.sort_values(["score", "return_pct"], ascending=False).head(n) def tag_rows(df: pd.DataFrame, source: str, family: str) -> pd.DataFrame: if df.empty: return df x = df.copy() x.insert(0, "source", source) x.insert(1, "family", family) return x def write_best(out: Path, name: str, df: pd.DataFrame, outputs: Dict[str, str], n: int = 25) -> pd.DataFrame: if df.empty: return df b = best_rows(df, n) p = out / name b.to_csv(p, index=False) outputs[name.removesuffix(".csv")] = str(p) return b def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--out-dir", default="DEX_REPORTS/oleg_live_preparation") ap.add_argument("--oleg-summary", default="DEX_REPORTS/oleg_tranche_dca_weth_usdc_existing_quarter_full/summary.csv") ap.add_argument("--oleg-deploy-fraction-summary", default="DEX_REPORTS/oleg_tranche_deploy_fraction_weth_usdc_existing_quarter/summary.csv") ap.add_argument("--revert-range-summary", default="DEX_REPORTS/revert_range_static_oor_weth_usdc_existing_quarter/summary.csv") ap.add_argument("--static-quarter", default="../DEX_full_cycle/DEX_REPORTS/weth_usdc_030_capacity_v1/summary_quarter.csv") ap.add_argument("--static-monthly-dir", default="../DEX_full_cycle/DEX_REPORTS/weth_usdc_030_capacity_v1") ap.add_argument("--pool-candidates", default="DEX_DATA/l2_pool_discovery/oleg_priority_candidates_2026_05_05.csv") args = ap.parse_args() out = Path(args.out_dir) out.mkdir(parents=True, exist_ok=True) oleg = read_csv(Path(args.oleg_summary)) oleg_deploy = read_csv(Path(args.oleg_deploy_fraction_summary)) revert_range = read_csv(Path(args.revert_range_summary)) static = read_csv(Path(args.static_quarter)) pools = read_csv(Path(args.pool_candidates)) outputs: Dict[str, str] = {} best_frames: List[pd.DataFrame] = [] for name, source, family, df in [ ("best_oleg_tranche_existing_weth.csv", "local", "oleg_tranche_full", oleg), ("best_oleg_deploy_fraction_existing_weth.csv", "local", "oleg_deploy_fraction", oleg_deploy), ("best_revert_range_existing_weth.csv", "local", "revert_range_static_oor", revert_range), ("best_static_existing_weth.csv", "DEX_full_cycle", "static_capacity", static), ]: b = write_best(out, name, tag_rows(df, source, family), outputs) if not b.empty: best_frames.append(b) monthly_best: List[pd.DataFrame] = [] monthly_dir = Path(args.static_monthly_dir) for period in ["feb", "mar", "apr", "febmar", "quarter"]: df = read_csv(monthly_dir / f"summary_{period}.csv") if df.empty: continue b = best_rows(tag_rows(df, "DEX_full_cycle", f"static_{period}"), 10) b.insert(2, "period", period) monthly_best.append(b) if monthly_best: m = pd.concat(monthly_best, ignore_index=True) p = out / "static_monthly_best_existing_weth.csv" m.to_csv(p, index=False) outputs["static_monthly_best"] = str(p) wf = read_csv(monthly_dir / "walkforward.csv") if not wf.empty: p = out / "static_walkforward_existing_weth.csv" wf.to_csv(p, index=False) outputs["static_walkforward"] = str(p) if best_frames: cmp = pd.concat(best_frames, ignore_index=True, sort=False) sort_cols = [c for c in ["score", "return_pct"] if c in cmp.columns] if sort_cols: cmp = cmp.sort_values(sort_cols, ascending=False) keep = [c for c in [ "source", "family", "strategy", "lower_pct", "upper_pct", "rebalance_mode", "rebalance_hours", "capital_usd", "total_capital_usd", "initial_deployed_usd", "initial_deploy_fraction", "parts", "return_pct", "return_on_total_capital_pct", "return_on_deployed_capital_pct", "return_on_initial_deployed_pct", "mdd_pct", "fees_earned_total", "fees_reinvested", "fees_uncollected_end", "inventory_pnl_ex_fees_usd", "fees_minus_inventory_loss_usd", "time_in_range_pct", "avg_liquidity_share_pct_when_in_range", "p95_liquidity_share_pct_when_in_range", "p99_liquidity_share_pct_when_in_range", "max_liquidity_share_pct_when_in_range", "rebalances", "valid_capacity_avg_p95_p99", "score", ] if c in cmp.columns] p = out / "combined_weth_live_candidates.csv" cmp[keep].to_csv(p, index=False) outputs["combined_weth_live_candidates"] = str(p) if not pools.empty: x = pools.copy() for col in ["tvl_usd", "volume_h24_usd", "volume_tvl_h24", "tx_h24", "score"]: if col in x.columns: x[col] = pd.to_numeric(x[col], errors="coerce") keep = [c for c in [ "network", "dex_name", "pool_name", "pool_address", "base_token_symbol", "quote_token_symbol", "tvl_usd", "volume_h24_usd", "volume_tvl_h24", "tx_h24", "score", "raw_api_url", ] if c in x.columns] ranked = x.sort_values(["score", "volume_tvl_h24", "volume_h24_usd"], ascending=False)[keep].head(50) p = out / "ranked_bluechip_pool_candidates.csv" ranked.to_csv(p, index=False) outputs["pool_candidates"] = str(p) summary: Dict[str, object] = { "outputs": outputs, "oleg_rows": int(len(oleg)), "oleg_deploy_fraction_rows": int(len(oleg_deploy)), "revert_range_rows": int(len(revert_range)), "static_rows": int(len(static)), "pool_candidate_rows": int(len(pools)), } (out / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") print(json.dumps(summary, ensure_ascii=False)) if __name__ == "__main__": main()