"""MYCO bonding surface CLI — run simulations from the terminal. Thin argparse wrapper calling existing simulation functions. Stdlib only. Usage: myco simulate token-launch --n-assets 3 --duration 90 myco compare-dca --total 10000 --chunks 20 myco signal-routing --trajectory volatile --steps 100 myco stress-test --fractions 0.02 0.05 0.10 0.20 """ import argparse import json import sys import math import numpy as np def _price_trajectory(name: str, steps: int) -> list[float]: """Generate a named price trajectory.""" t = np.linspace(0, 1, steps) if name == "stable": return [1.0] * steps elif name == "bull": return (1.0 + t * 0.5).tolist() elif name == "crash": return (1.0 - 0.4 * t + 0.1 * np.sin(t * 20)).tolist() elif name == "volatile": return (1.0 + 0.3 * np.sin(t * 30) + 0.1 * np.cos(t * 7)).tolist() else: raise ValueError(f"Unknown trajectory: {name}") def _save_chart(fig, path: str) -> None: """Save a matplotlib figure to disk.""" fig.savefig(path, dpi=150, bbox_inches="tight") print(f"Chart saved to {path}") def cmd_simulate(args: argparse.Namespace) -> int: """Run a simulation scenario.""" from src.composed.simulator import ( scenario_token_launch, scenario_bank_run, scenario_mixed_issuance, ) scenario_map = { "token-launch": lambda: scenario_token_launch( n_assets=args.n_assets, duration=args.duration, ), "bank-run": lambda: scenario_bank_run( n_assets=args.n_assets, duration=args.duration, ), "mixed-issuance": lambda: scenario_mixed_issuance( n_assets=args.n_assets, duration=args.duration, ), } if args.scenario not in scenario_map: print(f"Unknown scenario: {args.scenario}", file=sys.stderr) print(f"Available: {', '.join(scenario_map.keys())}", file=sys.stderr) return 1 result = scenario_map[args.scenario]() if args.output == "json": data = { k: v.tolist() for k, v in vars(result).items() } print(json.dumps(data, indent=2)) else: metrics = result print(f"Simulation: {args.scenario}") print(f" Duration: {args.duration}") print(f" Final supply: {metrics.supply[-1]:.2f}") print(f" Final reserve: {metrics.reserve_value[-1]:.2f}") print(f" Backing ratio: {metrics.backing_ratio[-1]:.4f}") print(f" Total minted: {metrics.financial_minted[-1] + metrics.commitment_minted[-1]:.2f}") print(f" Total redeemed: {metrics.total_redeemed[-1]:.2f}") if args.save_chart: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from src.utils.plotting import plot_time_series fig, axes = plt.subplots(2, 2, figsize=(12, 8)) plot_time_series(result.times, {"Supply": result.supply}, ax=axes[0, 0]) plot_time_series(result.times, {"Reserve": result.reserve_value}, ax=axes[0, 1]) plot_time_series(result.times, {"Backing Ratio": result.backing_ratio}, ax=axes[1, 0]) plot_time_series( result.times, {"Financial": result.financial_minted, "Commitment": result.commitment_minted}, ax=axes[1, 1], ) fig.suptitle(f"MYCO Simulation: {args.scenario}") fig.tight_layout() _save_chart(fig, args.save_chart) plt.close(fig) return 0 def cmd_compare_dca(args: argparse.Namespace) -> int: """Compare DCA strategies.""" from src.composed.simulator import scenario_dca_comparison results = scenario_dca_comparison( total_amount=args.total, n_chunks=args.chunks, interval=args.interval, ) if args.output == "json": data = {} for name, dca_result in results.items(): data[name] = { "total_tokens": dca_result.order.total_tokens_received, "total_spent": dca_result.order.total_spent, "avg_price": dca_result.order.avg_price, "twap_price": dca_result.twap_price, "lump_sum_tokens": dca_result.lump_sum_tokens, "dca_advantage": dca_result.dca_advantage, } print(json.dumps(data, indent=2)) else: for name, dca_result in results.items(): print(f"\nStrategy: {name}") print(f" Tokens received: {dca_result.order.total_tokens_received:.4f}") print(f" Avg price: {dca_result.order.avg_price:.6f}") print(f" TWAP price: {dca_result.twap_price:.6f}") print(f" Lump sum tokens: {dca_result.lump_sum_tokens:.4f}") print(f" DCA advantage: {dca_result.dca_advantage:+.4f}") if args.save_chart: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt fig, axes = plt.subplots(1, 2, figsize=(12, 5)) for name, dca_result in results.items(): history = dca_result.order.history chunks_x = list(range(len(history))) tokens = [h["tokens"] for h in history] prices = [h["price"] for h in history] axes[0].bar( [x + (0.2 if name == "twap_aware" else -0.2) for x in chunks_x], tokens, width=0.35, label=name, alpha=0.7, ) axes[1].plot(chunks_x, prices, "-o", label=name, markersize=4) axes[0].set_title("Tokens per Chunk") axes[0].set_xlabel("Chunk") axes[0].legend() axes[1].set_title("Price per Chunk") axes[1].set_xlabel("Chunk") axes[1].legend() fig.suptitle("DCA Strategy Comparison") fig.tight_layout() _save_chart(fig, args.save_chart) plt.close(fig) return 0 def cmd_signal_routing(args: argparse.Namespace) -> int: """Run signal routing simulation.""" from src.primitives.signal_router import ( AdaptiveParams, SignalRouterConfig, simulate_signal_routing, ) base = AdaptiveParams( flow_threshold=0.1, pamm_alpha_bar=10.0, surge_fee_rate=0.05, oracle_multiplier_velocity=0.0, ) config = SignalRouterConfig() prices = _price_trajectory(args.trajectory, args.steps) result = simulate_signal_routing(base, config, prices) if args.output == "json": print(json.dumps(result, indent=2)) else: print(f"Signal Routing: {args.trajectory} ({args.steps} steps)") print(f" Final flow_threshold: {result['flow_threshold'][-1]:.6f}") print(f" Final pamm_alpha_bar: {result['pamm_alpha_bar'][-1]:.4f}") print(f" Final surge_fee_rate: {result['surge_fee_rate'][-1]:.6f}") print(f" Final oracle_velocity: {result['oracle_velocity'][-1]:.6f}") print(f" Final volatility: {result['volatility'][-1]:.6f}") if args.save_chart: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt fig, axes = plt.subplots(3, 1, figsize=(12, 10), sharex=True) t = result["times"] axes[0].plot(t, prices, label="Spot Price") axes[0].set_ylabel("Price") axes[0].set_title(f"Signal Router: {args.trajectory}") axes[0].legend() axes[1].plot(t, result["twap_deviation"], label="TWAP Deviation") axes[1].plot(t, result["volatility"], label="Volatility") axes[1].set_ylabel("Signal Value") axes[1].legend() axes[2].plot(t, result["flow_threshold"], label="Flow Threshold") axes[2].plot(t, result["surge_fee_rate"], label="Surge Fee") axes[2].set_ylabel("Parameter Value") axes[2].set_xlabel("Time") axes[2].legend() fig.tight_layout() _save_chart(fig, args.save_chart) plt.close(fig) return 0 def cmd_stress_test(args: argparse.Namespace) -> int: """Run bank run stress tests at multiple redemption fractions.""" from src.composed.simulator import scenario_bank_run results = {} for frac in args.fractions: result = scenario_bank_run(redemption_fraction=frac) results[frac] = result final_ratio = result.backing_ratio[-1] final_reserve = result.reserve_value[-1] survived = "YES" if final_ratio > 0.5 else "NO" print(f" Fraction {frac:.2f}: backing={final_ratio:.4f} reserve={final_reserve:.0f} survived={survived}") if args.save_chart: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt fig, ax = plt.subplots(figsize=(10, 6)) for frac, result in results.items(): ax.plot(result.times, result.reserve_value, label=f"{frac:.0%} redemption") ax.set_xlabel("Time") ax.set_ylabel("Reserve Value") ax.set_title("Bank Run Stress Test — Reserve Curves") ax.legend() fig.tight_layout() _save_chart(fig, args.save_chart) plt.close(fig) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="myco", description="MYCO bonding surface simulation toolkit", ) sub = parser.add_subparsers(dest="command") # simulate p_sim = sub.add_parser("simulate", help="Run a simulation scenario") p_sim.add_argument("scenario", choices=["token-launch", "bank-run", "mixed-issuance"]) p_sim.add_argument("--n-assets", type=int, default=3) p_sim.add_argument("--duration", type=float, default=90.0) p_sim.add_argument("--output", choices=["text", "json"], default="text") p_sim.add_argument("--save-chart", type=str, default=None) # compare-dca p_dca = sub.add_parser("compare-dca", help="Compare DCA strategies") p_dca.add_argument("--total", type=float, default=10_000.0) p_dca.add_argument("--chunks", type=int, default=20) p_dca.add_argument("--interval", type=float, default=1.0) p_dca.add_argument("--output", choices=["text", "json"], default="text") p_dca.add_argument("--save-chart", type=str, default=None) # signal-routing p_sig = sub.add_parser("signal-routing", help="Signal routing simulation") p_sig.add_argument("--trajectory", choices=["stable", "bull", "crash", "volatile"], default="stable") p_sig.add_argument("--steps", type=int, default=100) p_sig.add_argument("--output", choices=["text", "json"], default="text") p_sig.add_argument("--save-chart", type=str, default=None) # stress-test p_stress = sub.add_parser("stress-test", help="Bank run stress tests") p_stress.add_argument( "--fractions", type=float, nargs="+", default=[0.02, 0.05, 0.10, 0.20], ) p_stress.add_argument("--save-chart", type=str, default=None) return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) if args.command is None: parser.print_help() return 0 handlers = { "simulate": cmd_simulate, "compare-dca": cmd_compare_dca, "signal-routing": cmd_signal_routing, "stress-test": cmd_stress_test, } return handlers[args.command](args) if __name__ == "__main__": sys.exit(main())