import pandas as pd
import numpy as np
from datetime import datetime
class SafeMomentumFilter:
    def __init__(self):
        self.volatility_regimes = {
            'NORMAL': (0.01, 0.02),      # 1-2% daily volatility
            'ELEVATED': (0.02, 0.03),    # 2-3% daily volatility
            'HIGH': (0.03, 0.05),        # 3-5% daily volatility
            'EXTREME': (0.05, 1.0)       # >5% daily volatility
        }
def identify_volatility_regime(self, prices, window=20):
        """
        Classify current volatility relative to historical norms
        """
returns = prices.pct_change()
        current_vol = returns.tail(20).std()
        historical_vol = returns.tail(252).std()
vol_ratio = current_vol / historical_vol
if vol_ratio < 0.8:
            return 'LOW', vol_ratio
        elif vol_ratio < 1.2:
            return 'NORMAL', vol_ratio
        elif vol_ratio < 1.5:
            return 'ELEVATED', vol_ratio
        elif vol_ratio < 2.0:
            return 'HIGH', vol_ratio
        else:
            return 'EXTREME', vol_ratio
def should_trade_by_volatility(self, regime, risk_tolerance='NORMAL'):
        """
        Trading safety rules based on volatility regime
        """
rules = {
            'CONSERVATIVE': {
                'LOW': True,
                'NORMAL': True,
                'ELEVATED': False,
                'HIGH': False,
                'EXTREME': False
            },
            'NORMAL': {
                'LOW': True,
                'NORMAL': True,
                'ELEVATED': True,
                'HIGH': False,
                'EXTREME': False
            },
            'AGGRESSIVE': {
                'LOW': True,
                'NORMAL': True,
                'ELEVATED': True,
                'HIGH': True,
                'EXTREME': False
            }
        }
return rules[risk_tolerance].get(regime, False)
Usage
prices = fetch_prices('AAPL', days=100)
regime, ratio = SafeMomentumFilter().identify_volatility_regime(prices)
can_trade = SafeMomentumFilter().should_trade_by_volatility(regime, risk_tolerance='NORMAL')
print(f"Volatility regime: {regime} ({ratio:.2f}x normal)")
print(f"Can trade: {can_trade}")

def measure_trend_maturity(prices, lookback=20):
    """
    Determine how far into a trend we are
    Early trend = safer to enter, Late trend = riskier
    """
# Trend length: consecutive closes above/below moving average
    sma = prices.rolling(window=lookback).mean()
    above_sma = prices > sma
consecutive_above = 0
    for i in range(len(above_sma)-1, -1, -1):
        if above_sma.iloc[i]:
            consecutive_above += 1
        else:
            break
# Calculate as percentage of lookback period
    trend_maturity = consecutive_above / lookback
return {
        'consecutive_bars': consecutive_above,
        'maturity_pct': trend_maturity,
        'safety_score': 1.0 - trend_maturity  # Lower is safer
    }
Safe to trade if trend < 50% complete (high safety score)
maturity = measure_trend_maturity(prices, lookback=20)
if maturity['maturity_pct'] < 0.5:
    print("Early-stage trend: SAFE TO TRADE")
elif maturity['maturity_pct'] < 0.75:
    print("Mid-stage trend: REDUCED SIZE")
else:
    print("Late-stage trend: AVOID OR SKIP")

def validate_liquidity_for_momentum(symbol, position_size_pct=0.1):
    """
    Ensure sufficient liquidity to exit if momentum breaks
    Can't hold large positions during low liquidity
    """
ticker = fetch_ticker(symbol)
    bid_ask_spread = ticker['ask'] - ticker['bid']
    spread_pct = bid_ask_spread / ticker['mid']
daily_volume = ticker['volume']
    position_notional = ticker['price']  position_size_pct  account_balance
# Position as percentage of daily volume
    position_as_pct_volume = position_notional / daily_volume
# Safety rules
    if spread_pct > 0.01:  # >1% spread
        return False, "WIDE_SPREAD"
    elif position_as_pct_volume > 0.05:  # Position > 5% of daily volume
        return False, "LOW_LIQUIDITY"
    else:
        return True, "LIQUID"
Usage
liquid, reason = validate_liquidity_for_momentum('AAPL', position_size_pct=0.05)
if liquid:
    execute_momentum_trade()
else:
    print(f"Insufficient liquidity: {reason}")

class SafePositionSizer:
    def __init__(self, account_balance, max_risk_per_trade=0.02, max_momentum_drawdown=0.05):
        self.balance = account_balance
        self.max_risk_per_trade = max_risk_per_trade  # 2% max loss per trade
        self.max_momentum_drawdown = max_momentum_drawdown  # 5% max account drawdown
        self.open_positions = {}
def calculate_safe_position_size(self, entry_price, stop_loss, atr, volatility_regime):
        """
        Position size adjusted for volatility and trend maturity
        """
# Base calculation
        risk_amount = self.balance * self.max_risk_per_trade
        stop_distance = abs(entry_price - stop_loss)
        base_size = risk_amount / stop_distance
# Volatility adjustment
        if volatility_regime == 'NORMAL':
            vol_multiplier = 1.0
        elif volatility_regime == 'ELEVATED':
            vol_multiplier = 0.7  # Reduce by 30%
        elif volatility_regime == 'HIGH':
            vol_multiplier = 0.4  # Reduce by 60%
        else:  # EXTREME
            return 0  # Don't trade
# Trend maturity adjustment
        maturity = measure_trend_maturity(prices)
        if maturity['maturity_pct'] < 0.3:
            maturity_multiplier = 1.0  # Full size early trend
        elif maturity['maturity_pct'] < 0.5:
            maturity_multiplier = 0.85
        elif maturity['maturity_pct'] < 0.75:
            maturity_multiplier = 0.5
        else:
            return 0  # Don't trade mature trend
# Final size
        final_size = base_size  vol_multiplier  maturity_multiplier
# Portfolio-level constraint
        portfolio_exposure = sum([p['size'] * p['price'] for p in self.open_positions.values()])
        if (portfolio_exposure + final_size  entry_price) > (self.balance  2.0):
            # Cap at 2x leverage
            final_size = (self.balance * 2.0 - portfolio_exposure) / entry_price
return max(0, final_size)
def check_account_level_safety(self):
        """
        Circuit breaker: stop trading if account drawdown > threshold
        """
current_equity = self.calculate_portfolio_value()
        daily_drawdown = (self.balance - current_equity) / self.balance
if daily_drawdown > 0.05:  # Lost 5% today
            return False, "DAILY_LOSS_LIMIT_HIT"
worst_open_loss = min([p['current_pnl'] for p in self.open_positions.values()])
        if worst_open_loss < -(self.balance * 0.03):  # Single position down 3%
            return False, "SINGLE_POSITION_LIMIT"
return True, "SAFE"

def advanced_stop_loss_strategy(entry_price, atr, signal_type='BUY',
                                volatility_regime='NORMAL', trend_maturity=0.3):
    """
    Adaptive stop loss that accounts for market conditions
    """
if volatility_regime == 'NORMAL':
        stop_multiplier = 2.0  # 2x ATR
    elif volatility_regime == 'ELEVATED':
        stop_multiplier = 2.5  # 2.5x ATR
    elif volatility_regime == 'HIGH':
        stop_multiplier = 3.0  # 3x ATR
    else:
        stop_multiplier = 4.0  # Don't trade, but if forced: 4x ATR
# Tighten stops in mature trends (more prone to reversal)
    if trend_maturity > 0.75:
        stop_multiplier *= 0.7  # 30% tighter
stop_distance = stop_multiplier * atr
if signal_type == 'BUY':
        stop_loss = entry_price - stop_distance
    else:  # SHORT
        stop_loss = entry_price + stop_distance
return {
        'stop_loss': stop_loss,
        'stop_distance_pips': stop_distance,
        'stop_pct': (stop_distance / entry_price) * 100
    }
def trailing_stop_for_momentum(entry_price, current_price, atr, trail_distance=2.0):
    """
    Trailing stop: protects profits as price rises
    Moves up but never down
    """
highest_price = max(entry_price, current_price)
    trail_amount = trail_distance * atr
trailing_stop = highest_price - trail_amount
# Safety: never move stop into loss
    if trailing_stop < entry_price:
        trailing_stop = entry_price
return trailing_stop