Equity Curve Visualisation

# Install necessary libraries: pandas for data manipulation, numpy for numerical operations,
# and plotly for interactive visualizations.
!pip install pandas numpy plotly

import warnings
warnings.filterwarnings("ignore") # Suppress warnings for cleaner output.

import pandas as pd # Import pandas for data analysis and manipulation.
import numpy as np  # Import numpy for numerical operations, especially array handling.
import plotly.graph_objects as go # Import plotly.graph_objects for creating rich interactive charts.
from plotly.subplots import make_subplots # Import make_subplots to create multi-panel figures.

def generate_data(periods: int) -> pd.DataFrame:
    """Generates synthetic price data for simulation purposes."""
    start_date     = pd.to_datetime("2024-01-01 00:00:00+00:00")
    datetime_index = pd.date_range(start_date, periods=periods, freq="1min", tz="UTC")
    price_data = []; last_close = 42000 # Initial closing price for simulation.
    volatility_scale = 0.005; wick_scale = 0.002 # Scaling factors for price movement.

    for _ in range(periods):
        # Simulate open, close, high, and low prices based on normal distribution.
        open_price  = last_close + np.random.normal(0, last_close * volatility_scale * 0.1)
        close_price = open_price + np.random.normal(0, last_close * volatility_scale)
        body_high   = max(open_price, close_price)
        body_low    = min(open_price, close_price)
        high_price  = max(body_high + abs(np.random.normal(0, last_close * wick_scale)),
                          open_price, close_price)
        low_price   = min(body_low  - abs(np.random.normal(0, last_close * wick_scale)),
                          open_price, close_price)

        # Ensure high is always greater than or equal to low.
        if high_price < low_price:
            high_price, low_price = low_price, high_price

        # Append generated OHLCV data.
        price_data.append({
            "open":  max(1, int(open_price)),
            "high":  max(1, int(high_price)),
            "low":   max(1, int(low_price)),
            "close": max(1, int(close_price)),
        })
        last_close = close_price

    # Create DataFrame from simulated data.
    df = pd.DataFrame(price_data, index=datetime_index)
    df.index.name = "datetime"
    df["volume"]   = np.random.uniform(100.0, 500.0, periods) # Simulate trading volume.
    df["datetime"] = df.index.to_series() # Add datetime as a column.
    return df.reset_index(drop=True)

# Generate 500 periods of synthetic data.
df = generate_data(500)
display(df.head()) # Display the first 5 rows of the generated DataFrame.

INITIAL_CAPITAL  = 10_000.0 # Starting capital for the strategy simulation.
FAST_WINDOW      = 10     # Number of periods for the fast moving average.
SLOW_WINDOW      = 30     # Number of periods for the slow moving average.
FEE_PCT          = 0.0005 # Transaction fee percentage per trade (0.05%).

# Calculate fast and slow moving averages.
df["fast_ma"]          = df["close"].rolling(FAST_WINDOW).mean()
df["slow_ma"]          = df["close"].rolling(SLOW_WINDOW).mean()

# Generate trading signal: 1 for buy (fast_ma > slow_ma), 0 otherwise.
df["signal"]           = np.where(df["fast_ma"] > df["slow_ma"], 1, 0)

# Determine position based on previous period's signal, filling initial NaN with 0.
df["position"]         = df["signal"].shift(1).fillna(0)

# Identify trade events (changes in position).
df["trade"]            = df["position"].diff().abs()

# Calculate market returns (percentage change in close price).
df["market_return"]    = df["close"].pct_change()

# Calculate strategy returns, subtracting fees for each trade.
df["strategy_return"]  = df["position"] * df["market_return"] - df["trade"] * FEE_PCT

# Drop rows with NaN values introduced by rolling windows and shift operations.
df = df.dropna().reset_index(drop=True)

# Calculate cumulative equity for the strategy and benchmark (buy-and-hold).
df["strategy_equity"]  = INITIAL_CAPITAL * (1 + df["strategy_return"]).cumprod()
df["market_equity"]    = INITIAL_CAPITAL * (1 + df["market_return"]).cumprod()

# Calculate drawdown for strategy and benchmark.
df["drawdown"]         = (df["strategy_equity"] / df["strategy_equity"].cummax() - 1) * 100
df["market_drawdown"]  = (df["market_equity"]   / df["market_equity"].cummax()   - 1) * 100

# Identify buy and sell entry points for visualization.
buy_entries  = df[(df["position"] == 1) & (df["position"].shift(1) == 0)]
sell_entries = df[(df["position"] == 0) & (df["position"].shift(1) == 1)]

# --- Summary performance statistics ---
total_ret  = (df["strategy_equity"].iloc[-1] / INITIAL_CAPITAL - 1) * 100
mkt_ret    = (df["market_equity"].iloc[-1]   / INITIAL_CAPITAL - 1) * 100
max_dd     = df["drawdown"].min()
sharpe     = (df["strategy_return"].mean() / df["strategy_return"].std()) * np.sqrt(525_600) # Annualized Sharpe ratio assuming 1-minute data (525,600 minutes/year).
n_trades   = int(df["trade"].sum())

print("--- Equity Curve Summary ---")
print(f"  Initial Capital  : ${INITIAL_CAPITAL:,.2f}")
print(f"  Final Equity     : ${df['strategy_equity'].iloc[-1]:,.2f}")
print(f"  Strategy Return  : {total_ret:+.2f}%")
print(f"  Benchmark Return : {mkt_ret:+.2f}%")
print(f"  Max Drawdown     : {max_dd:.2f}%")
print(f"  Sharpe Ratio     : {sharpe:.4f}")
print(f"  Total Trade Events: {n_trades}")

fig = make_subplots(
    rows=4, cols=1,
    shared_xaxes=True, # Ensure all subplots share the same x-axis.
    subplot_titles=[   # Titles for each subplot panel.
        "Candlestick Chart + MA Crossover + Trade Signals",
        "Equity Curve — Strategy vs Buy-and-Hold",
        "Drawdown (%)",
        "Position Exposure + Per-Period Returns",
    ],
    row_heights=[0.35, 0.30, 0.20, 0.15], # Proportional heights for each row.
    vertical_spacing=0.04, # Spacing between subplots.
)

# --- Panel 1: Price chart with Moving Averages and Trade Signals ---
# Candlestick chart to visualize price action.
fig.add_trace(go.Candlestick(
    x=df["datetime"],
    open=df["open"], high=df["high"],
    low=df["low"],   close=df["close"],
    name="Price"), row=1, col=1)

# Fast Moving Average line.
fig.add_trace(go.Scatter(
    x=df["datetime"], y=df["fast_ma"],
    mode="lines", name=f"Fast MA ({FAST_WINDOW})",
    line=dict(color="blue",   width=1)), row=1, col=1)

# Slow Moving Average line.
fig.add_trace(go.Scatter(
    x=df["datetime"], y=df["slow_ma"],
    mode="lines", name=f"Slow MA ({SLOW_WINDOW})",
    line=dict(color="orange", width=1)), row=1, col=1)

# Add markers for buy entries (green triangles).
if len(buy_entries) > 0:
    fig.add_trace(go.Scatter(
        x=buy_entries["datetime"],
        y=buy_entries["low"] * 0.999, # Place marker slightly below the low price.
        mode="markers",
        marker=dict(symbol="triangle-up",   size=10, color="green"),
        name="Buy Entry"), row=1, col=1)

# Add markers for sell exits (red triangles).
if len(sell_entries) > 0:
    fig.add_trace(go.Scatter(
        x=sell_entries["datetime"],
        y=sell_entries["high"] * 1.001, # Place marker slightly above the high price.
        mode="markers",
        marker=dict(symbol="triangle-down", size=10, color="red"),
        name="Sell Exit"), row=1, col=1)

# --- Panel 2: Equity Curves (Strategy vs Buy-and-Hold) ---
# Strategy equity curve.
fig.add_trace(go.Scatter(
    x=df["datetime"], y=df["strategy_equity"],
    mode="lines", name="Strategy",
    line=dict(color="green", width=2)), row=2, col=1)

# Benchmark (buy-and-hold) equity curve.
fig.add_trace(go.Scatter(
    x=df["datetime"], y=df["market_equity"],
    mode="lines", name="Buy-and-Hold",
    line=dict(color="gray", width=1.5, dash="dash")), row=2, col=1)

# Annotation for the final strategy equity value.
fig.add_annotation(
    x=df["datetime"].iloc[-1],
    y=df["strategy_equity"].iloc[-1],
    text=f"${df['strategy_equity'].iloc[-1]:,.0f}",
    showarrow=True, arrowhead=2, arrowcolor="green",
    font=dict(color="green"),
    row=2, col=1,
)

# --- Panel 3: Drawdown ---
# Strategy drawdown curve.
fig.add_trace(go.Scatter(
    x=df["datetime"], y=df["drawdown"],
    mode="lines", name="Strategy DD (%)",
    fill="tozeroy", # Fill area below the curve.
    line=dict(color="red", width=1)), row=3, col=1)

# Benchmark drawdown curve.
fig.add_trace(go.Scatter(
    x=df["datetime"], y=df["market_drawdown"],
    mode="lines", name="Market DD (%)",
    line=dict(color="orange", width=1, dash="dot")), row=3, col=1)

# Vertical line annotation for maximum drawdown point.
max_dd_idx = df["drawdown"].idxmin()
fig.add_vline(
    x=df["datetime"].iloc[max_dd_idx].timestamp() * 1000, # Convert Timestamp to Unix milliseconds for Plotly.
    line_dash="dash", line_color="darkred",
    annotation_text=f"MDD: {max_dd:.1f}%", # Display max drawdown value.
    row=3, col=1,
)

# --- Panel 4: Position Exposure and Per-Period Returns ---
# Bar chart showing when the strategy holds a position.
fig.add_trace(go.Bar(
    x=df["datetime"], y=df["position"],
    name="In Position",
    marker_color="rgba(0,128,0,0.3)"), row=4, col=1)

# Line chart showing per-period returns.
fig.add_trace(go.Scatter(
    x=df["datetime"], y=df["strategy_return"] * 100,
    mode="lines", name="Return (%)",
    line=dict(color="purple", width=0.8), yaxis="y5"), row=4, col=1)

# --- Layout and Axis Configuration ---
fig.update_layout(
    title_text=( # Main title for the entire figure, including key performance metrics.
        f"Full Equity Curve Report — MA Crossover Strategy  |  "
        f"Return: {total_ret:+.2f}%  |  MDD: {max_dd:.2f}%  |  Sharpe: {sharpe:.2f}"
    ),
    xaxis_rangeslider_visible=False, # Hide the range slider for a cleaner look.
    height=1100, # Overall height of the figure.
    yaxis_title="Price", # Y-axis title for the first panel.
    yaxis2_title="Portfolio ($)", # Y-axis title for the second panel.
    yaxis3_title="Drawdown (%)", # Y-axis title for the third panel.
    yaxis4_title="Position", # Y-axis title for the fourth panel.
    yaxis5=dict(title="Return (%)", overlaying="y4", side="right"), # Secondary Y-axis for returns on panel 4.
    xaxis4_title="Datetime", # X-axis title for the bottom-most panel.
    legend=dict(orientation="h", y=-0.05), # Horizontal legend placed below the plot.
)
fig.show()