scripts/drawdown_analyzer.py
#!/usr/bin/env python3
"""Equity curve drawdown analysis with response recommendations.
Analyzes an equity curve to identify all drawdown periods, calculate
maximum drawdown statistics, and provide actionable recommendations
based on the current drawdown state.
Usage:
python scripts/drawdown_analyzer.py --demo
python scripts/drawdown_analyzer.py --equity equity_data.json
Dependencies:
uv pip install numpy
Environment Variables:
None required.
"""
import argparse
import json
import sys
from dataclasses import dataclass
from typing import Optional
import numpy as np
# ── Data Models ─────────────────────────────────────────────────────
@dataclass
class DrawdownPeriod:
"""A single drawdown period from peak to recovery."""
start_index: int
trough_index: int
recovery_index: Optional[int] # None if not yet recovered
peak_value: float
trough_value: float
depth: float # As positive fraction (0.10 = 10%)
duration_to_trough: int # Periods from start to trough
recovery_duration: Optional[int] # Periods from trough to recovery
total_duration: Optional[int] # Periods from start to recovery
@dataclass
class DrawdownSummary:
"""Summary statistics for all drawdowns in an equity curve."""
max_drawdown: float
max_drawdown_period: Optional[DrawdownPeriod]
current_drawdown: float
current_drawdown_start: Optional[int]
total_time_underwater: int
longest_underwater: int
num_drawdowns: int
avg_drawdown_depth: float
avg_recovery_time: float
all_periods: list[DrawdownPeriod]
# ── Core Analysis ───────────────────────────────────────────────────
def find_drawdown_periods(
equity: np.ndarray, min_depth: float = 0.01
) -> list[DrawdownPeriod]:
"""Identify all drawdown periods in an equity curve.
Args:
equity: Array of equity values over time.
min_depth: Minimum drawdown depth to record (default 1%).
Returns:
List of DrawdownPeriod objects sorted by start index.
"""
if len(equity) < 2:
return []
periods: list[DrawdownPeriod] = []
peak = equity[0]
peak_index = 0
in_drawdown = False
dd_start = 0
trough = equity[0]
trough_index = 0
for i in range(len(equity)):
if equity[i] >= peak:
# New high or recovery
if in_drawdown:
depth = (peak - trough) / peak if peak > 0 else 0.0
if depth >= min_depth:
periods.append(DrawdownPeriod(
start_index=dd_start,
trough_index=trough_index,
recovery_index=i,
peak_value=peak,
trough_value=trough,
depth=depth,
duration_to_trough=trough_index - dd_start,
recovery_duration=i - trough_index,
total_duration=i - dd_start,
))
in_drawdown = False
peak = equity[i]
peak_index = i
trough = equity[i]
trough_index = i
else:
if not in_drawdown:
in_drawdown = True
dd_start = peak_index
trough = equity[i]
trough_index = i
if equity[i] < trough:
trough = equity[i]
trough_index = i
# Handle open drawdown (not yet recovered)
if in_drawdown:
depth = (peak - trough) / peak if peak > 0 else 0.0
if depth >= min_depth:
periods.append(DrawdownPeriod(
start_index=dd_start,
trough_index=trough_index,
recovery_index=None,
peak_value=peak,
trough_value=trough,
depth=depth,
duration_to_trough=trough_index - dd_start,
recovery_duration=None,
total_duration=None,
))
return periods
def analyze_drawdowns(equity: np.ndarray) -> DrawdownSummary:
"""Comprehensive drawdown analysis of an equity curve.
Args:
equity: Array of equity values over time.
Returns:
DrawdownSummary with all statistics.
"""
periods = find_drawdown_periods(equity)
# Maximum drawdown
max_dd = 0.0
max_dd_period: Optional[DrawdownPeriod] = None
for p in periods:
if p.depth > max_dd:
max_dd = p.depth
max_dd_period = p
# Current drawdown
peak = np.max(equity)
current = equity[-1]
current_dd = (peak - current) / peak if peak > 0 else 0.0
current_dd_start: Optional[int] = None
if current_dd > 0.001:
# Find the latest all-time high before the current underwater period.
# np.argmax returns the first max, which can overstate drawdown duration
# after an equity curve retests the same peak and then sells off again.
peak_indices = np.flatnonzero(equity == peak)
current_dd_start = int(peak_indices[-1])
# Time underwater
peaks = np.maximum.accumulate(equity)
underwater = peaks > equity
total_underwater = int(np.sum(underwater))
# Longest consecutive underwater period
longest_uw = 0
current_uw = 0
for uw in underwater:
if uw:
current_uw += 1
longest_uw = max(longest_uw, current_uw)
else:
current_uw = 0
# Average drawdown depth
avg_depth = np.mean([p.depth for p in periods]) if periods else 0.0
# Average recovery time (only for recovered drawdowns)
recovered = [p for p in periods if p.recovery_duration is not None]
avg_recovery = (
np.mean([p.recovery_duration for p in recovered]) if recovered else 0.0
)
return DrawdownSummary(
max_drawdown=max_dd,
max_drawdown_period=max_dd_period,
current_drawdown=current_dd,
current_drawdown_start=current_dd_start,
total_time_underwater=total_underwater,
longest_underwater=longest_uw,
num_drawdowns=len(periods),
avg_drawdown_depth=float(avg_depth),
avg_recovery_time=float(avg_recovery),
all_periods=periods,
)
def recovery_required(drawdown: float) -> float:
"""Calculate gain needed to recover from a drawdown.
Args:
drawdown: Drawdown as positive fraction (0.20 = 20%).
Returns:
Required gain as positive fraction.
"""
if drawdown >= 1.0:
return float("inf")
if drawdown <= 0.0:
return 0.0
return drawdown / (1.0 - drawdown)
def drawdown_response(drawdown: float) -> tuple[str, str, str]:
"""Determine the appropriate response for a given drawdown level.
Returns:
Tuple of (level, status_color, recommendation).
"""
if drawdown < 0.05:
return ("Normal", "\033[92m", "Continue trading at full size.")
elif drawdown < 0.10:
return ("Caution", "\033[93m", "Reduce position sizes by 25-50%. Review recent trades for errors.")
elif drawdown < 0.15:
return ("Warning", "\033[91m", "Minimum position sizes only. Review strategy edge and market regime.")
elif drawdown < 0.20:
return ("Critical", "\033[91m", "Halt new trades. Manage existing positions only. Mandatory review.")
else:
return (
"Emergency",
"\033[91m",
"Full stop. Close positions systematically. 48-72 hour break. "
"Complete strategy review before resuming.",
)
# ── Output Formatting ──────────────────────────────────────────────
def print_separator(char: str = "=", width: int = 70) -> None:
"""Print a separator line."""
print(char * width)
def print_summary(summary: DrawdownSummary, equity: np.ndarray) -> None:
"""Print formatted drawdown analysis results."""
reset = "\033[0m"
print()
print_separator()
print(" DRAWDOWN ANALYSIS")
print_separator()
print(f"\n Equity Curve: {len(equity)} periods")
print(f" Start Value: {equity[0]:.2f}")
print(f" End Value: {equity[-1]:.2f}")
print(f" Peak Value: {np.max(equity):.2f}")
print(f" Total Return: {(equity[-1] / equity[0] - 1) * 100:+.1f}%")
# ── Maximum Drawdown ────────────────────────────────────────
print()
print_separator("-")
print(" MAXIMUM DRAWDOWN")
print_separator("-")
print(f" Max Drawdown: {summary.max_drawdown:.1%}")
print(f" Recovery Required: +{recovery_required(summary.max_drawdown):.1%}")
if summary.max_drawdown_period:
p = summary.max_drawdown_period
print(f" Peak Index: {p.start_index}")
print(f" Trough Index: {p.trough_index}")
print(f" Peak Value: {p.peak_value:.2f}")
print(f" Trough Value: {p.trough_value:.2f}")
print(f" Duration to Trough: {p.duration_to_trough} periods")
if p.recovery_index is not None:
print(f" Recovery Index: {p.recovery_index}")
print(f" Recovery Duration: {p.recovery_duration} periods")
print(f" Total Duration: {p.total_duration} periods")
else:
print(" Recovery: NOT YET RECOVERED")
# ── Current Status ──────────────────────────────────────────
print()
print_separator("-")
print(" CURRENT STATUS")
print_separator("-")
level, color, recommendation = drawdown_response(summary.current_drawdown)
print(f" Current Drawdown: {color}{summary.current_drawdown:.1%}{reset}")
print(f" Status Level: {color}{level}{reset}")
print(f" Recovery Needed: +{recovery_required(summary.current_drawdown):.1%}")
print(f" Recommendation: {recommendation}")
if summary.current_drawdown_start is not None and summary.current_drawdown > 0.001:
periods_in_dd = len(equity) - 1 - summary.current_drawdown_start
print(f" Periods in Current Drawdown: {periods_in_dd}")
# ── Underwater Analysis ─────────────────────────────────────
print()
print_separator("-")
print(" UNDERWATER ANALYSIS")
print_separator("-")
total_periods = len(equity)
uw_pct = summary.total_time_underwater / total_periods * 100 if total_periods > 0 else 0
print(f" Total Time Underwater: {summary.total_time_underwater} periods ({uw_pct:.1f}%)")
print(f" Longest Underwater: {summary.longest_underwater} periods")
print(f" Number of Drawdowns (>1%): {summary.num_drawdowns}")
print(f" Average Drawdown Depth: {summary.avg_drawdown_depth:.1%}")
if summary.avg_recovery_time > 0:
print(f" Average Recovery Time: {summary.avg_recovery_time:.1f} periods")
# ── All Drawdown Periods ────────────────────────────────────
if summary.all_periods:
print()
print_separator("-")
print(" ALL DRAWDOWN PERIODS")
print_separator("-")
print(f" {'#':>3s} {'Depth':>7s} {'Peak':>8s} {'Trough':>8s} {'To Trough':>10s} {'Recovery':>10s} {'Status':<12s}")
print(" " + "-" * 65)
for i, p in enumerate(sorted(summary.all_periods, key=lambda x: -x.depth), 1):
recovery_str = (
f"{p.recovery_duration}" if p.recovery_duration is not None else "OPEN"
)
status_str = "Recovered" if p.recovery_index is not None else "ACTIVE"
print(
f" {i:>3d} {p.depth:>6.1%} {p.peak_value:>8.2f} {p.trough_value:>8.2f} "
f"{p.duration_to_trough:>10d} {recovery_str:>10s} {status_str:<12s}"
)
# ── Recovery Table ──────────────────────────────────────────
print()
print_separator("-")
print(" RECOVERY REFERENCE TABLE")
print_separator("-")
print(f" {'Drawdown':>10s} {'Gain Needed':>12s} {'At 1%/day':>10s}")
print(" " + "-" * 36)
for dd_pct in [5, 10, 15, 20, 25, 30, 40, 50]:
dd = dd_pct / 100
gain = recovery_required(dd)
days = 0
cumulative = 1.0
target = 1.0 / (1.0 - dd)
while cumulative < target and days < 1000:
cumulative *= 1.01
days += 1
print(f" {dd:>9.0%} {gain:>11.1%} {days:>8d} days")
print()
print_separator()
# ── Demo Data ───────────────────────────────────────────────────────
def generate_demo_equity(
start: float = 100.0,
periods: int = 200,
seed: int = 42,
) -> np.ndarray:
"""Generate a realistic equity curve with multiple drawdowns.
Creates an equity curve that trends upward with realistic drawdown
characteristics including:
- A moderate drawdown early on (~8%)
- A significant drawdown in the middle (~18%)
- A recovery followed by a mild current drawdown (~6%)
Args:
start: Starting equity value.
periods: Number of periods to generate.
seed: Random seed for reproducibility.
Returns:
NumPy array of equity values.
"""
rng = np.random.default_rng(seed)
equity = [start]
current = start
# Phase 1: Mild uptrend (periods 0-40)
for _ in range(40):
ret = rng.normal(0.003, 0.015)
current *= (1 + ret)
equity.append(current)
# Phase 2: Moderate drawdown (periods 41-60)
for _ in range(20):
ret = rng.normal(-0.004, 0.012)
current *= (1 + ret)
equity.append(current)
# Phase 3: Recovery and new highs (periods 61-100)
for _ in range(40):
ret = rng.normal(0.004, 0.014)
current *= (1 + ret)
equity.append(current)
# Phase 4: Significant drawdown (periods 101-130)
for _ in range(30):
ret = rng.normal(-0.006, 0.015)
current *= (1 + ret)
equity.append(current)
# Phase 5: Slow recovery (periods 131-170)
for _ in range(40):
ret = rng.normal(0.005, 0.013)
current *= (1 + ret)
equity.append(current)
# Phase 6: Current mild drawdown (periods 171-200)
for _ in range(periods - 171):
ret = rng.normal(-0.001, 0.012)
current *= (1 + ret)
equity.append(current)
return np.array(equity[:periods])
def load_equity_from_file(filepath: str) -> np.ndarray:
"""Load equity curve from a JSON file.
Expected format: {"equity": [100.0, 101.5, 99.8, ...]}
Or a plain JSON array: [100.0, 101.5, 99.8, ...]
"""
try:
with open(filepath, "r") as f:
data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"Error loading equity file: {e}")
sys.exit(1)
if isinstance(data, list):
return np.array(data, dtype=float)
elif isinstance(data, dict) and "equity" in data:
return np.array(data["equity"], dtype=float)
else:
print("Expected JSON array or object with 'equity' key")
sys.exit(1)
# ── Main ────────────────────────────────────────────────────────────
def main() -> None:
"""Entry point for the drawdown analyzer."""
parser = argparse.ArgumentParser(
description="Equity curve drawdown analysis — identify and analyze drawdown periods"
)
parser.add_argument(
"--demo",
action="store_true",
help="Run with generated demo equity curve",
)
parser.add_argument(
"--equity",
type=str,
help="Path to JSON file with equity curve data",
)
parser.add_argument(
"--min-depth",
type=float,
default=0.02,
help="Minimum drawdown depth to report (default: 0.02 = 2%%)",
)
args = parser.parse_args()
if args.demo:
equity = generate_demo_equity()
print("\n [Running with generated demo equity curve]")
elif args.equity:
equity = load_equity_from_file(args.equity)
else:
parser.print_help()
print("\nProvide --demo or --equity <file.json>")
sys.exit(1)
summary = analyze_drawdowns(equity)
# Re-run with custom min_depth if specified
if args.min_depth != 0.01:
summary.all_periods = find_drawdown_periods(equity, min_depth=args.min_depth)
summary.num_drawdowns = len(summary.all_periods)
if summary.all_periods:
summary.avg_drawdown_depth = float(
np.mean([p.depth for p in summary.all_periods])
)
print_summary(summary, equity)
if __name__ == "__main__":
main()
scripts/risk_dashboard.py
#!/usr/bin/env python3
"""Portfolio risk dashboard with limit checking and color-coded status.
Analyzes a portfolio of positions against configurable risk limits and
displays a comprehensive dashboard showing exposure, concentration,
drawdown, and circuit breaker status.
Usage:
python scripts/risk_dashboard.py --demo
python scripts/risk_dashboard.py --positions positions.json
Dependencies:
None (pure Python, no external packages required)
Environment Variables:
ACCOUNT_SIZE: Total account size in SOL (default: 100)
"""
import argparse
import json
import math
import os
import sys
from dataclasses import dataclass, field
from typing import Optional
# ── Configuration ───────────────────────────────────────────────────
ACCOUNT_SIZE = float(os.getenv("ACCOUNT_SIZE", "100"))
# Risk limits (configurable)
LIMITS = {
"max_single_position_pct": 0.10, # 10% of account
"max_total_exposure_pct": 0.80, # 80% of account
"max_daily_loss_pct": 0.03, # 3% daily loss
"max_drawdown_warning_pct": 0.10, # 10% drawdown warning
"max_drawdown_critical_pct": 0.15, # 15% drawdown critical
"max_drawdown_halt_pct": 0.20, # 20% drawdown halt
"max_consecutive_losses": 3, # consecutive loss warning
"max_sector_concentration_pct": 0.30, # 30% per sector
"max_concurrent_positions": 10, # maximum open positions
}
# ── Data Models ─────────────────────────────────────────────────────
@dataclass
class Position:
"""A single portfolio position."""
token: str
entry_price: float
current_price: float
size_sol: float
stop_loss: Optional[float] = None
sector: str = "unknown"
token_type: str = "mid-cap" # blue-chip, mid-cap, small-cap, micro, pumpfun
@property
def pnl_sol(self) -> float:
"""Unrealized P&L in SOL."""
if self.entry_price == 0:
return 0.0
return self.size_sol * (self.current_price / self.entry_price - 1.0)
@property
def pnl_pct(self) -> float:
"""Unrealized P&L as percentage."""
if self.entry_price == 0:
return 0.0
return (self.current_price / self.entry_price - 1.0) * 100
@property
def risk_to_stop(self) -> float:
"""Risk in SOL if stop loss is hit."""
if self.stop_loss is None or self.entry_price == 0:
return self.size_sol # Assume 100% loss if no stop
loss_pct = (self.entry_price - self.stop_loss) / self.entry_price
return self.size_sol * max(0.0, loss_pct)
@property
def current_value(self) -> float:
"""Current position value in SOL."""
if self.entry_price == 0:
return 0.0
return self.size_sol * (self.current_price / self.entry_price)
@dataclass
class PortfolioState:
"""Aggregate portfolio state for risk assessment."""
account_size: float
positions: list[Position]
realized_pnl_today: float = 0.0
equity_peak: float = 0.0
consecutive_losses: int = 0
consecutive_wins: int = 0
trade_history: list[float] = field(default_factory=list)
# ── Status Helpers ──────────────────────────────────────────────────
class Status:
"""Color-coded status indicators."""
OK = "OK"
WARNING = "WARNING"
BREACH = "BREACH"
def colorize(text: str, status: str) -> str:
"""Add ANSI color codes based on status."""
colors = {
Status.OK: "\033[92m", # Green
Status.WARNING: "\033[93m", # Yellow
Status.BREACH: "\033[91m", # Red
}
reset = "\033[0m"
color = colors.get(status, "")
return f"{color}{text}{reset}"
def status_icon(status: str) -> str:
"""Return a text icon for the status."""
icons = {
Status.OK: "[OK]",
Status.WARNING: "[WARN]",
Status.BREACH: "[BREACH]",
}
return icons.get(status, "[??]")
# ── Risk Calculations ───────────────────────────────────────────────
def calculate_total_exposure(positions: list[Position], account_size: float) -> tuple[float, float]:
"""Calculate total deployed capital.
Returns:
Tuple of (total_sol, percentage_of_account).
"""
total = sum(p.size_sol for p in positions)
pct = total / account_size if account_size > 0 else 0.0
return total, pct
def calculate_total_risk(positions: list[Position], account_size: float) -> tuple[float, float]:
"""Calculate total portfolio risk (distance to stops).
Returns:
Tuple of (total_risk_sol, percentage_of_account).
"""
total = sum(p.risk_to_stop for p in positions)
pct = total / account_size if account_size > 0 else 0.0
return total, pct
def calculate_largest_position(positions: list[Position], account_size: float) -> tuple[str, float, float]:
"""Find the largest single position.
Returns:
Tuple of (token_name, size_sol, percentage_of_account).
"""
if not positions:
return ("none", 0.0, 0.0)
largest = max(positions, key=lambda p: p.size_sol)
pct = largest.size_sol / account_size if account_size > 0 else 0.0
return (largest.token, largest.size_sol, pct)
def calculate_hhi(positions: list[Position]) -> float:
"""Calculate Herfindahl-Hirschman Index for position concentration.
Returns:
HHI value between 0 and 10000.
- 10000: single position (maximum concentration)
- <1500: well diversified
- 1500-2500: moderate concentration
- >2500: high concentration
"""
if not positions:
return 0.0
total = sum(p.size_sol for p in positions)
if total == 0:
return 0.0
shares = [(p.size_sol / total) * 100 for p in positions]
return sum(s * s for s in shares)
def calculate_daily_pnl(
positions: list[Position], realized_pnl: float, account_size: float
) -> tuple[float, float]:
"""Calculate total daily P&L (realized + unrealized).
Returns:
Tuple of (total_pnl_sol, percentage_of_account).
"""
unrealized = sum(p.pnl_sol for p in positions)
total = realized_pnl + unrealized
pct = total / account_size if account_size > 0 else 0.0
return total, pct
def calculate_drawdown(current_equity: float, equity_peak: float) -> float:
"""Calculate current drawdown from peak.
Returns:
Drawdown as a positive fraction (0.10 = 10% drawdown).
"""
if equity_peak <= 0:
return 0.0
return max(0.0, (equity_peak - current_equity) / equity_peak)
def calculate_sector_concentration(
positions: list[Position], account_size: float
) -> dict[str, float]:
"""Calculate allocation percentage per sector.
Returns:
Dict mapping sector name to percentage of account.
"""
sectors: dict[str, float] = {}
for p in positions:
sectors[p.sector] = sectors.get(p.sector, 0.0) + p.size_sol
return {s: v / account_size for s, v in sectors.items()} if account_size > 0 else {}
def recovery_needed(drawdown: float) -> float:
"""Calculate the gain needed to recover from a drawdown.
Args:
drawdown: Drawdown as a positive fraction (e.g., 0.20 for 20%).
Returns:
Required gain as a positive fraction.
"""
if drawdown >= 1.0:
return float("inf")
if drawdown <= 0.0:
return 0.0
return drawdown / (1.0 - drawdown)
def drawdown_response_level(drawdown: float) -> tuple[str, str]:
"""Determine drawdown response level and recommendation.
Returns:
Tuple of (level_name, recommendation).
"""
if drawdown < 0.05:
return ("Normal", "Continue trading at full size")
elif drawdown < 0.10:
return ("Caution", "Reduce position sizes by 25-50%")
elif drawdown < 0.15:
return ("Warning", "Minimum position sizes only")
elif drawdown < 0.20:
return ("Critical", "Halt new trades, manage existing only")
else:
return ("Emergency", "Full stop, review everything before resuming")
# ── Dashboard Output ────────────────────────────────────────────────
def print_separator(char: str = "=", width: int = 70) -> None:
"""Print a separator line."""
print(char * width)
def print_header(title: str) -> None:
"""Print a section header."""
print()
print_separator()
print(f" {title}")
print_separator()
def print_metric(
label: str, value: str, status: str, limit_desc: str = ""
) -> None:
"""Print a single metric line with status."""
icon = colorize(status_icon(status), status)
limit_text = f" (limit: {limit_desc})" if limit_desc else ""
print(f" {icon} {label:<35s} {value:<20s}{limit_text}")
def run_dashboard(state: PortfolioState) -> dict[str, str]:
"""Run the risk dashboard and print results.
Args:
state: Current portfolio state.
Returns:
Dict of check names to status values for programmatic use.
"""
results: dict[str, str] = {}
limits = LIMITS
print_header("PORTFOLIO RISK DASHBOARD")
print(f" Account Size: {state.account_size:.2f} SOL")
print(f" Open Positions: {len(state.positions)}")
print(f" Equity Peak: {state.equity_peak:.2f} SOL")
# ── Exposure ────────────────────────────────────────────────
print_header("EXPOSURE")
total_sol, total_pct = calculate_total_exposure(state.positions, state.account_size)
exp_status = (
Status.BREACH if total_pct > limits["max_total_exposure_pct"]
else Status.WARNING if total_pct > limits["max_total_exposure_pct"] * 0.8
else Status.OK
)
print_metric(
"Total Exposure",
f"{total_sol:.2f} SOL ({total_pct:.1%})",
exp_status,
f"{limits['max_total_exposure_pct']:.0%}",
)
results["total_exposure"] = exp_status
cash = state.account_size - total_sol
cash_pct = cash / state.account_size if state.account_size > 0 else 0
cash_status = Status.OK if cash_pct >= 0.20 else Status.WARNING if cash_pct >= 0.10 else Status.BREACH
print_metric("Cash Reserve", f"{cash:.2f} SOL ({cash_pct:.1%})", cash_status, ">= 20%")
results["cash_reserve"] = cash_status
pos_count_status = (
Status.BREACH if len(state.positions) > limits["max_concurrent_positions"]
else Status.WARNING if len(state.positions) > limits["max_concurrent_positions"] * 0.8
else Status.OK
)
print_metric(
"Concurrent Positions",
f"{len(state.positions)}",
pos_count_status,
f"<= {limits['max_concurrent_positions']}",
)
results["concurrent_positions"] = pos_count_status
# ── Concentration ───────────────────────────────────────────
print_header("CONCENTRATION")
token_name, token_sol, token_pct = calculate_largest_position(state.positions, state.account_size)
pos_status = (
Status.BREACH if token_pct > limits["max_single_position_pct"]
else Status.WARNING if token_pct > limits["max_single_position_pct"] * 0.8
else Status.OK
)
print_metric(
f"Largest Position ({token_name})",
f"{token_sol:.2f} SOL ({token_pct:.1%})",
pos_status,
f"{limits['max_single_position_pct']:.0%}",
)
results["largest_position"] = pos_status
hhi = calculate_hhi(state.positions)
hhi_status = Status.OK if hhi < 1500 else Status.WARNING if hhi < 2500 else Status.BREACH
hhi_label = "Low" if hhi < 1500 else "Moderate" if hhi < 2500 else "High"
print_metric("Concentration (HHI)", f"{hhi:.0f} ({hhi_label})", hhi_status, "< 2500")
results["hhi"] = hhi_status
sectors = calculate_sector_concentration(state.positions, state.account_size)
for sector, pct in sorted(sectors.items(), key=lambda x: -x[1]):
sec_status = (
Status.BREACH if pct > limits["max_sector_concentration_pct"]
else Status.WARNING if pct > limits["max_sector_concentration_pct"] * 0.8
else Status.OK
)
print_metric(
f" Sector: {sector}",
f"{pct:.1%}",
sec_status,
f"{limits['max_sector_concentration_pct']:.0%}",
)
results[f"sector_{sector}"] = sec_status
# ── Risk ────────────────────────────────────────────────────
print_header("RISK")
risk_sol, risk_pct = calculate_total_risk(state.positions, state.account_size)
risk_status = Status.OK if risk_pct < 0.05 else Status.WARNING if risk_pct < 0.10 else Status.BREACH
print_metric("Portfolio Risk (to stops)", f"{risk_sol:.2f} SOL ({risk_pct:.1%})", risk_status, "< 10%")
results["portfolio_risk"] = risk_status
# ── Daily P&L ───────────────────────────────────────────────
print_header("DAILY P&L")
daily_sol, daily_pct = calculate_daily_pnl(
state.positions, state.realized_pnl_today, state.account_size
)
daily_status = (
Status.BREACH if daily_pct < -limits["max_daily_loss_pct"]
else Status.WARNING if daily_pct < -limits["max_daily_loss_pct"] * 0.5
else Status.OK
)
pnl_sign = "+" if daily_sol >= 0 else ""
print_metric(
"Daily P&L",
f"{pnl_sign}{daily_sol:.2f} SOL ({pnl_sign}{daily_pct:.1%})",
daily_status,
f"> -{limits['max_daily_loss_pct']:.0%}",
)
results["daily_pnl"] = daily_status
# ── Drawdown ────────────────────────────────────────────────
print_header("DRAWDOWN")
current_equity = state.account_size + sum(p.pnl_sol for p in state.positions) + state.realized_pnl_today
dd = calculate_drawdown(current_equity, state.equity_peak) if state.equity_peak > 0 else 0.0
dd_level, dd_rec = drawdown_response_level(dd)
dd_status = (
Status.BREACH if dd >= limits["max_drawdown_critical_pct"]
else Status.WARNING if dd >= limits["max_drawdown_warning_pct"]
else Status.OK
)
print_metric("Current Drawdown", f"{dd:.1%} ({dd_level})", dd_status, f"< {limits['max_drawdown_warning_pct']:.0%}")
results["drawdown"] = dd_status
if dd > 0:
rec = recovery_needed(dd)
print_metric("Recovery Needed", f"+{rec:.1%}", Status.WARNING if dd >= 0.10 else Status.OK)
print(f" Recommendation: {dd_rec}")
# ── Streaks ─────────────────────────────────────────────────
print_header("STREAKS & CIRCUIT BREAKERS")
if state.consecutive_losses > 0:
streak_status = (
Status.BREACH if state.consecutive_losses >= 5
else Status.WARNING if state.consecutive_losses >= limits["max_consecutive_losses"]
else Status.OK
)
print_metric(
"Consecutive Losses",
f"{state.consecutive_losses}",
streak_status,
f"< {limits['max_consecutive_losses']}",
)
results["consecutive_losses"] = streak_status
if state.consecutive_losses >= 7:
print(f" ACTION: Halt trading for 24 hours, full review required")
elif state.consecutive_losses >= 5:
print(f" ACTION: Minimum position sizes only")
elif state.consecutive_losses >= 3:
print(f" ACTION: Reduce position sizes by 50%")
else:
print_metric("Consecutive Losses", "0", Status.OK, f"< {limits['max_consecutive_losses']}")
results["consecutive_losses"] = Status.OK
if state.consecutive_wins > 0:
print_metric("Consecutive Wins", f"{state.consecutive_wins}", Status.OK)
# ── Positions Detail ────────────────────────────────────────
if state.positions:
print_header("POSITION DETAILS")
print(f" {'Token':<12s} {'Size':>8s} {'Entry':>10s} {'Current':>10s} {'P&L':>10s} {'P&L%':>8s} {'Risk':>8s}")
print(" " + "-" * 68)
for p in sorted(state.positions, key=lambda x: -x.size_sol):
pnl_sign = "+" if p.pnl_sol >= 0 else ""
print(
f" {p.token:<12s} {p.size_sol:>7.2f}S {p.entry_price:>10.6f} "
f"{p.current_price:>10.6f} {pnl_sign}{p.pnl_sol:>8.2f}S "
f"{pnl_sign}{p.pnl_pct:>6.1f}% {p.risk_to_stop:>7.2f}S"
)
# ── Summary ─────────────────────────────────────────────────
print_header("SUMMARY")
breaches = [k for k, v in results.items() if v == Status.BREACH]
warnings = [k for k, v in results.items() if v == Status.WARNING]
if breaches:
print(colorize(f" BREACHES ({len(breaches)}):", Status.BREACH))
for b in breaches:
print(colorize(f" - {b}", Status.BREACH))
if warnings:
print(colorize(f" WARNINGS ({len(warnings)}):", Status.WARNING))
for w in warnings:
print(colorize(f" - {w}", Status.WARNING))
if not breaches and not warnings:
print(colorize(" All checks passed. Portfolio within risk limits.", Status.OK))
print()
return results
# ── Demo Data ───────────────────────────────────────────────────────
def create_demo_portfolio() -> PortfolioState:
"""Create a realistic demo portfolio for dashboard demonstration."""
positions = [
Position(
token="SOL",
entry_price=145.00,
current_price=142.50,
size_sol=8.0,
stop_loss=135.00,
sector="infrastructure",
token_type="blue-chip",
),
Position(
token="JUP",
entry_price=0.85,
current_price=0.92,
size_sol=5.0,
stop_loss=0.75,
sector="defi",
token_type="large-cap",
),
Position(
token="RAY",
entry_price=2.10,
current_price=1.95,
size_sol=4.0,
stop_loss=1.80,
sector="defi",
token_type="large-cap",
),
Position(
token="BONK",
entry_price=0.00002,
current_price=0.000025,
size_sol=3.0,
stop_loss=0.000015,
sector="meme",
token_type="mid-cap",
),
Position(
token="WIF",
entry_price=1.80,
current_price=1.65,
size_sol=2.5,
stop_loss=1.50,
sector="meme",
token_type="mid-cap",
),
Position(
token="NEWMEME",
entry_price=0.001,
current_price=0.0008,
size_sol=0.5,
stop_loss=None, # No stop on PumpFun
sector="meme",
token_type="pumpfun",
),
Position(
token="ORCA",
entry_price=3.50,
current_price=3.60,
size_sol=3.0,
stop_loss=3.10,
sector="defi",
token_type="large-cap",
),
]
return PortfolioState(
account_size=ACCOUNT_SIZE,
positions=positions,
realized_pnl_today=-0.8,
equity_peak=ACCOUNT_SIZE * 1.05, # Was 5% higher at peak
consecutive_losses=2,
consecutive_wins=0,
)
def load_positions_from_file(filepath: str) -> PortfolioState:
"""Load portfolio state from a JSON file.
Expected format:
{
"account_size": 100,
"equity_peak": 105,
"realized_pnl_today": -0.5,
"consecutive_losses": 1,
"consecutive_wins": 0,
"positions": [
{
"token": "SOL",
"entry_price": 145.0,
"current_price": 142.5,
"size_sol": 8.0,
"stop_loss": 135.0,
"sector": "infrastructure",
"token_type": "blue-chip"
}
]
}
"""
try:
with open(filepath, "r") as f:
data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"Error loading positions file: {e}")
sys.exit(1)
positions = []
for p in data.get("positions", []):
positions.append(Position(
token=p["token"],
entry_price=p["entry_price"],
current_price=p["current_price"],
size_sol=p["size_sol"],
stop_loss=p.get("stop_loss"),
sector=p.get("sector", "unknown"),
token_type=p.get("token_type", "mid-cap"),
))
return PortfolioState(
account_size=data.get("account_size", ACCOUNT_SIZE),
positions=positions,
realized_pnl_today=data.get("realized_pnl_today", 0.0),
equity_peak=data.get("equity_peak", data.get("account_size", ACCOUNT_SIZE)),
consecutive_losses=data.get("consecutive_losses", 0),
consecutive_wins=data.get("consecutive_wins", 0),
)
# ── Main ────────────────────────────────────────────────────────────
def main() -> None:
"""Entry point for the risk dashboard."""
parser = argparse.ArgumentParser(
description="Portfolio risk dashboard — analyze positions against risk limits"
)
parser.add_argument(
"--demo",
action="store_true",
help="Run with demo portfolio data",
)
parser.add_argument(
"--positions",
type=str,
help="Path to JSON file with portfolio positions",
)
args = parser.parse_args()
if args.demo:
state = create_demo_portfolio()
print("\n [Running with demo portfolio data]")
elif args.positions:
state = load_positions_from_file(args.positions)
else:
parser.print_help()
print("\nProvide --demo or --positions <file.json>")
sys.exit(1)
results = run_dashboard(state)
# Exit with non-zero code if any breaches
breaches = [k for k, v in results.items() if v == Status.BREACH]
sys.exit(1 if breaches else 0)
if __name__ == "__main__":
main()