Risk Management

from finlab import data
from finlab.backtest import sim

data.set_market('us')
with data.us_universe(index='SP500'):
    close = data.get('price:adj_close')
position = close > close.average(20)

# Single holding limit at 10%
report = sim(
    position,
    resample='M',
    position_limit=0.1,  # Any single holding capped at 10% of total capital
)

# Method 1: Use is_largest to limit holdings count
top_n_position = (close > close.average(20)).is_largest(30)

# Method 2: Limit during backtesting
report = sim(
    position,
    resample='M',
    position_limit=0.1,
    # Caps each holding at 10% of the portfolio
)

# Calculate volatility
returns = close.pct_change()
volatility = returns.rolling(60).std()

# Higher volatility = smaller position
inverse_vol_weight = 1 / volatility
normalized_weight = inverse_vol_weight / inverse_vol_weight.sum(axis=1)

# Combine with stock selection signal
position_with_vol = (close > close.average(20)) * normalized_weight

# Entry above the 20-day average, exit below the 60-day average
entry_signal = close > close.average(20)
exit_signal = close < close.average(60)

# Stop loss at 10%, take profit at 20%
position = entry_signal.hold_until(exit=exit_signal, stop_loss=0.10, take_profit=0.20)

report = sim(position, resample='M')

# Run initial backtest without stops
position_without_stops = entry_signal.hold_until(exit=exit_signal)
report_initial = sim(position_without_stops, resample='M', upload=False)

# Examine MAE/MFE
trades = report_initial.get_trades()
mae_q75 = abs(trades['mae'].quantile(0.75))
gmfe_q75 = trades['gmfe'].quantile(0.75)

print(f"Recommended stop loss: {mae_q75 * 1.2 * 100:.1f}%")  # MAE Q75 * 1.2
print(f"Recommended take profit: {gmfe_q75 * 0.8 * 100:.1f}%")  # GMFE Q75 * 0.8

# Use optimized stop loss/take profit
position_optimized = entry_signal.hold_until(
    exit=exit_signal, stop_loss=mae_q75 * 1.2, take_profit=gmfe_q75 * 0.8
)

# Signal exit after an 8 percentage-point drop from peak return
position = entry_signal.hold_until(exit=exit_signal, stop_loss=0.10)

report = sim(position, resample='M', trail_stop=0.08)

import pandas as pd

# Auto-exit after holding for 30 trading days
def add_time_stop(position, max_hold_days=30):
    """Add a time-based stop"""
    held = position > 0
    held_so_far = held.cumsum()
    # Consecutive days held: count since the last day without the position
    hold_days = held_so_far - held_so_far.where(~held).ffill().fillna(0)
    time_exit = hold_days > max_hold_days
    return position & (~time_exit)

position_with_time_stop = add_time_stop(position, max_hold_days=30)

def kelly_criterion(win_rate, avg_win, avg_loss):
    """
    Calculate Kelly fraction

    Args:
        win_rate: win rate (0-1)
        avg_win: average win rate
        avg_loss: average loss rate (positive value)

    Returns:
        float: recommended investment fraction
    """
    if avg_loss == 0:
        return 0

    kelly = (win_rate * avg_win - (1 - win_rate) * avg_loss) / avg_win
    return max(0, min(kelly, 1))  # Clamp to 0-1

# Usage example
stats = report.get_stats()
trades = report.get_trades()
kelly_ratio = kelly_criterion(
    win_rate=stats['win_ratio'],
    avg_win=trades[trades['return'] > 0]['return'].mean(),
    avg_loss=abs(trades[trades['return'] < 0]['return'].mean()),
)

print(f"Kelly recommended allocation: {kelly_ratio * 100:.1f}%")
# In practice, use 0.5 * Kelly (half-Kelly) for a more conservative approach
print(f"Half-Kelly recommendation: {kelly_ratio * 0.5 * 100:.1f}%")

# Invest a fixed fraction of total capital each time
RISK_PER_TRADE = 0.02  # 2% risk per trade

# Calculate position size based on stop loss level
stop_loss = 0.10  # 10% stop loss
position_size = RISK_PER_TRADE / stop_loss  # = 0.2 = 20%

print(f"Recommended position size: {position_size * 100:.0f}%")

def check_drawdown_limit(report, max_drawdown=0.20):
    """Check if maximum drawdown limit is exceeded"""
    stats = report.get_stats()
    current_dd = stats['max_drawdown']

    if abs(current_dd) > max_drawdown:
        print(
            f"Warning: drawdown {current_dd * 100:.1f}% exceeds limit {max_drawdown * 100:.0f}%"
        )
        print("Recommend pausing trading and reviewing the strategy!")
        return False
    return True

# Usage
if not check_drawdown_limit(report, max_drawdown=0.25):
    # Pause trading logic
    pass

def adjust_position_by_drawdown(position, creturn, base_size=1.0):
    """Dynamically adjust positions based on current drawdown

    Larger drawdown = smaller positions
    """
    # Calculate current drawdown
    peak = creturn.expanding().max()
    drawdown = (creturn - peak) / peak

    # Adjustment factor (at -20% drawdown, positions halved)
    adj_factor = 1 + drawdown / 0.20
    adj_factor = adj_factor.clip(0.5, 1.0)  # Clamp to 0.5-1.0

    # Adjust positions
    adjusted_position = position * adj_factor.reindex(position.index, method='ffill')
    return adjusted_position

# Usage
position_adjusted = adjust_position_by_drawdown(position, report.creturn, base_size=1.0)

def risk_alert(report):
    """Risk alert check"""
    stats = report.get_stats()
    alerts = []

    # Check 1: Sharpe ratio too low
    if stats['daily_sharpe'] < 0.5:
        alerts.append(f"⚠️ Sharpe ratio too low: {stats['daily_sharpe']:.2f}")

    # Check 2: Max drawdown too large
    if abs(stats['max_drawdown']) > 0.30:
        alerts.append(f"⚠️ Max drawdown too large: {stats['max_drawdown'] * 100:.1f}%")

    # Check 3: Win rate too low
    if stats['win_ratio'] < 0.40:
        alerts.append(f"⚠️ Win rate too low: {stats['win_ratio'] * 100:.1f}%")

    # Check 4: Average profit/loss ratio too low
    trades = report.get_trades()
    avg_win = trades[trades['return'] > 0]['return'].mean()
    avg_loss = abs(trades[trades['return'] < 0]['return'].mean())
    win_loss_ratio = avg_win / avg_loss if avg_loss > 0 else 0

    if win_loss_ratio < 1.5:
        alerts.append(f"⚠️ Win/loss ratio too low: {win_loss_ratio:.2f}")

    # Output alerts
    if alerts:
        print("=" * 50)
        print("Risk alerts:")
        for alert in alerts:
            print(alert)
        print("=" * 50)
    else:
        print("✅ Risk check passed")

    return len(alerts) == 0

# Usage
risk_alert(report)

def generate_risk_report(report):
    """Generate risk report"""
    stats = report.get_stats()
    trades = report.get_trades()

    risk_metrics = {
        'Annualized return': f"{stats['cagr'] * 100:.2f}%",
        'Annualized volatility': f"{stats['daily_vol'] * 100:.2f}%",
        'Sharpe ratio': f"{stats['daily_sharpe']:.2f}",
        'Max drawdown': f"{stats['max_drawdown'] * 100:.2f}%",
        'Win rate': f"{stats['win_ratio'] * 100:.1f}%",
        'Avg holding days': f"{trades['period'].mean():.1f} days",
        'Total trades': len(trades),
    }

    print("\n" + "=" * 50)
    print("Risk Report")
    print("=" * 50)
    for key, value in risk_metrics.items():
        print(f"{key:24s}: {value}")
    print("=" * 50 + "\n")

# Usage
generate_risk_report(report)

from finlab import data
from finlab.backtest import sim

# Load data
data.set_market('us')
with data.us_universe(index='SP500'):
    close = data.get('price:adj_close')

# Strategy logic
entry = close > close.average(20)
exit_signal = close < close.average(10)

# Add risk controls
position = entry.hold_until(
    exit=exit_signal,
    stop_loss=0.10,  # Stop loss 10%
    take_profit=0.20,  # Take profit 20%
)

# Backtest (single holding limit 10%, trailing stop 8%)
report = sim(position, resample='M', position_limit=0.1, trail_stop=0.08, upload=False)

# Risk check
if risk_alert(report):
    print("Strategy passed risk check; consider live trading")
else:
    print("Strategy failed risk check; optimization needed")

# Generate risk report
generate_risk_report(report)