Controles de riesgo y cumplimiento normativo

Por qué los agentes financieros necesitan controles estrictos

Límites del tamaño de las posiciones

MAX_POSITION_PCT = 0.20   # 20% of portfolio max per asset

def check_position_size(ticker: str, trade_usd: float,
                        portfolio_value: float) -> dict:
    proposed_pct = trade_usd / portfolio_value
    if proposed_pct > MAX_POSITION_PCT:
        return {
            'allowed': False,
            'reason': (
                f'Position size {proposed_pct:.1%} exceeds limit {MAX_POSITION_PCT:.1%}. '
                f'Max trade for this portfolio: USD {portfolio_value * MAX_POSITION_PCT:.0f}'
            )
        }
    return {'allowed': True, 'position_pct': round(proposed_pct, 4)}

if __name__ == '__main__':
    print('Large trade:', check_position_size('TSLA', 30000, 100000))
    print('Small trade:', check_position_size('TSLA', 10000, 100000))

Lista de instrumentos prohibidos

PROHIBITED_TICKERS = {
    'UVXY',   # VIX leveraged ETF — too volatile
    'TVIX',   # Delisted leveraged VIX
    'JDST',   # 3x leveraged gold miners short
    'NUGT',   # 3x leveraged gold miners long
    'YANG',   # 3x China short
}

PROHIBITED_TYPES = {'penny_stock', 'leveraged_3x', 'inverse_volatility'}

def check_prohibited_instrument(ticker: str) -> dict:
    if ticker.upper() in PROHIBITED_TICKERS:
        return {
            'allowed': False,
            'reason': f'{ticker} is on the prohibited instruments list'
        }
    return {'allowed': True}

if __name__ == '__main__':
    print("Checking UVXY:", check_prohibited_instrument('UVXY'))
    print("Checking AAPL:", check_prohibited_instrument('AAPL'))

Comprobación de la regla Pattern Day Trader

from datetime import datetime, timedelta, timezone

def check_pdt_rule(account_value_usd: float,
                   recent_trades: list[dict]) -> dict:
    if account_value_usd >= 25_000:
        return {'at_risk': False, 'reason': 'Account above PDT threshold'}

    cutoff = datetime.now(timezone.utc) - timedelta(days=5)
    recent = [
        t for t in recent_trades
        if datetime.fromisoformat(t['timestamp']) > cutoff
        and t.get('is_day_trade', False)
    ]

    if len(recent) >= 3:
        return {
            'at_risk': True,
            'day_trades_in_5_days': len(recent),
            'reason': (
                f'Account has {len(recent)} day trades in the last 5 days. '
                f'A 4th would trigger the PDT restriction.'
            )
        }
    return {'at_risk': False, 'day_trades_in_5_days': len(recent)}

if __name__ == '__main__':
    now = datetime.now(timezone.utc)
    trades = [
        {'timestamp': (now - timedelta(days=1)).isoformat(), 'is_day_trade': True},
        {'timestamp': (now - timedelta(days=2)).isoformat(), 'is_day_trade': True},
        {'timestamp': (now - timedelta(days=3)).isoformat(), 'is_day_trade': True},
    ]
    result = check_pdt_rule(account_value_usd=10000, recent_trades=trades)
    print('PDT rule check:', result)

Escalado del tamaño de las operaciones según la volatilidad

import yfinance as yf, numpy as np

BASELINE_VOL = 0.15   # 15% annualized vol — S&P 500 average

def volatility_scaled_size(ticker: str, base_allocation_usd: float) -> dict:
    hist = yf.Ticker(ticker).history(period='1y')['Close'].pct_change().dropna()
    vol = float(hist.std() * np.sqrt(252))

    scaling_factor = BASELINE_VOL / max(vol, 0.01)
    adjusted_usd   = base_allocation_usd * scaling_factor

    return {
        'ticker':             ticker,
        'annualized_vol':     round(vol, 4),
        'scaling_factor':     round(scaling_factor, 3),
        'adjusted_size_usd':  round(adjusted_usd, 2),
        'note': 'High-vol asset: position scaled down' if scaling_factor < 1 else 'Normal sizing'
    }

Registro de auditoría de cada recomendación

import json, hashlib, time, logging

audit_logger = logging.getLogger('financial_agent.audit')
audit_logger.setLevel(logging.INFO)

def audit_log(user_id: str, action: str, ticker: str,
              amount_usd: float, guardrail_results: list[dict],
              approved: bool, reason: str):
    log_entry = {
        'timestamp':         time.time(),
        'user_id':           user_id,
        'action':            action,
        'ticker':            ticker,
        'amount_usd':        amount_usd,
        'guardrails':        guardrail_results,
        'approved':          approved,
        'reason':            reason,
        'entry_id':          hashlib.md5(f'{user_id}{time.time()}'.encode()).hexdigest()[:12]
    }
    audit_logger.info(json.dumps(log_entry))

if __name__ == '__main__':
    import sys
    audit_logger.addHandler(logging.StreamHandler(sys.stdout))
    audit_log('user_42', 'BUY', 'AAPL', 5000.0,
               [{'check': 'position_size', 'allowed': True}],
               True, 'Within limits')

Inyección de avisos de exención de responsabilidad

FINANCIAL_DISCLAIMER = (
    '\n\n---\n'
    'DISCLAIMER: This output is generated by an AI agent and is for '
    'informational purposes only. It does not constitute personalized financial advice. '
    'Past performance does not guarantee future results. '
    'Please consult a licensed financial advisor before making investment decisions.'
)

def inject_disclaimer(response: str) -> str:
    if 'DISCLAIMER' not in response:
        return response + FINANCIAL_DISCLAIMER
    return response

if __name__ == '__main__':
    response = 'AAPL is currently trading at $150 with strong upward momentum.'
    print(inject_disclaimer(response))

Ejecución secuencial de todos los controles

def guardrail_check(user_id: str, ticker: str, action: str,
                    amount_usd: float, portfolio_value: float,
                    account_value: float, recent_trades: list[dict]) -> dict:
    checks = []

    # 1. Prohibited instruments
    prohibited = check_prohibited_instrument(ticker)
    checks.append({'check': 'prohibited_instrument', **prohibited})
    if not prohibited['allowed']:
        audit_log(user_id, action, ticker, amount_usd, checks, False, prohibited['reason'])
        return {'approved': False, 'reason': prohibited['reason'], 'checks': checks}

    # 2. Position size
    size_check = check_position_size(ticker, amount_usd, portfolio_value)
    checks.append({'check': 'position_size', **size_check})
    if not size_check['allowed']:
        audit_log(user_id, action, ticker, amount_usd, checks, False, size_check['reason'])
        return {'approved': False, 'reason': size_check['reason'], 'checks': checks}

    # 3. PDT rule
    if action == 'day_trade':
        pdt = check_pdt_rule(account_value, recent_trades)
        checks.append({'check': 'pdt_rule', **pdt})
        if pdt['at_risk']:
            return {'approved': False, 'reason': pdt['reason'], 'checks': checks}

    audit_log(user_id, action, ticker, amount_usd, checks, True, 'All guardrails passed')
    return {'approved': True, 'checks': checks}

Comprobación de idoneidad según el perfil de riesgo

RISK_PROFILES = {
    'conservative':   {'max_beta': 0.8,  'max_vol_pct': 15},
    'moderate':       {'max_beta': 1.2,  'max_vol_pct': 25},
    'aggressive':     {'max_beta': 2.0,  'max_vol_pct': 50},
}

def suitability_check(risk_profile: str, ticker_beta: float,
                       ticker_vol_pct: float) -> dict:
    limits = RISK_PROFILES.get(risk_profile, RISK_PROFILES['moderate'])
    if ticker_beta > limits['max_beta']:
        return {
            'suitable': False,
            'reason': f'Beta {ticker_beta:.2f} exceeds {risk_profile} limit {limits["max_beta"]}'
        }
    if ticker_vol_pct > limits['max_vol_pct']:
        return {
            'suitable': False,
            'reason': f'Volatility {ticker_vol_pct:.1f}% exceeds {risk_profile} limit {limits["max_vol_pct"]}%'
        }
    return {'suitable': True}

if __name__ == '__main__':
    print('Conservative check:', suitability_check('conservative', ticker_beta=1.5, ticker_vol_pct=20))
    print('Aggressive check:  ', suitability_check('aggressive', ticker_beta=1.5, ticker_vol_pct=20))

Limitación de la frecuencia de las recomendaciones

import time

last_recommendation: dict[str, float] = {}  # ticker -> timestamp
COOLDOWN_HOURS = 24

def check_recommendation_cooldown(ticker: str) -> dict:
    last = last_recommendation.get(ticker, 0)
    elapsed_hours = (time.time() - last) / 3600

    if elapsed_hours < COOLDOWN_HOURS:
        remaining = COOLDOWN_HOURS - elapsed_hours
        return {
            'allowed': False,
            'reason': f'Recommendation cooldown: {remaining:.1f} hours remaining for {ticker}'
        }
    return {'allowed': True}

def record_recommendation(ticker: str):
    last_recommendation[ticker] = time.time()

if __name__ == '__main__':
    print('Before recommending:', check_recommendation_cooldown('AAPL'))
    record_recommendation('AAPL')
    print('Right after recommending:', check_recommendation_cooldown('AAPL'))

Disyuntores de stop-loss y pérdida máxima

MAX_DAILY_DRAWDOWN_PCT = 0.05   # 5% daily loss limit

def check_daily_drawdown(portfolio_start_of_day: float,
                          portfolio_current: float) -> dict:
    daily_loss = (portfolio_current - portfolio_start_of_day) / portfolio_start_of_day
    if daily_loss < -MAX_DAILY_DRAWDOWN_PCT:
        return {
            'circuit_breaker': True,
            'daily_loss_pct':  round(daily_loss * 100, 2),
            'reason': (
                f'Portfolio down {abs(daily_loss)*100:.1f}% today. '
                f'Agent suspended — human review required.'
            )
        }
    return {'circuit_breaker': False, 'daily_loss_pct': round(daily_loss * 100, 2)}

if __name__ == '__main__':
    print('Small loss day:', check_daily_drawdown(100000, 98000))
    print('Big loss day:  ', check_daily_drawdown(100000, 92000))

¿Qué restringe la regla Pattern Day Trader (PDT)?

Resumen de los controles de riesgo y cumplimiento