AI Agents · 课时

风险与合规护栏

Q: 「风险与合规护栏」这节课中我会学到什么？

仓位规模限制、禁止交易工具检查和审计日志记录 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

仓位规模限制、禁止交易工具检查和审计日志记录

第 3 / 4 课13 个步骤

风险与合规护栏 是 CoddyKit 上的免费 AI Agents 课时。这是第 3 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

为什么金融智能体需要硬性防护栏

如果金融智能体可以不受限制地推荐任何交易，就可能让用户面临灾难性损失或监管违规。与人类顾问不同，智能体会始终如一地执行其推荐逻辑——这意味着一个错误或配置问题会影响所有用户。

硬性防护栏可以防止最严重的后果。

持仓规模限制

限制智能体建议投入任何单项资产的投资组合比例。常见规则是：任何单项持仓都不应超过投资组合总价值的 20%。这可以防止灾难性的集中风险。

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))

禁止交易的工具清单

维护一份智能体不得推荐的工具清单：低价股、高度杠杆化的交易型开放式指数基金、受制裁证券，或不适合散户投资者的工具。

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'))

日内交易者规则检查

在 US，如果账户资产低于 USD 25,000，并且在 5 个工作日内完成至少 4 次往返交易，该账户就会被标记为日内交易者（PDT）。智能体在推荐当日交易前必须检查交易频率。

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)

基于波动率缩放交易规模

波动性越高的资产，风险越大。请根据波动率反向调整持仓规模：如果某项资产的波动率是基准的两倍，就建议将持仓规模减半。

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'
    }

记录每一项推荐以便审计

每一项交易建议——无论被接受还是被拒绝——都必须记录完整上下文：时间戳、用户 ID、建议采取的行动、已执行的防护栏检查以及结果。这会为合规审查建立完整的审计轨迹。

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')

注入免责声明

智能体生成的任何金融输出都必须包含免责声明，说明这不是个性化的金融建议。请自动注入免责声明，绝不要依赖 LLM 始终如一地添加它。

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))

按顺序运行所有防护栏

创建一个单独的 guardrail_check() 函数，按顺序运行所有检查，并返回带有原因的通过/失败决定。智能体在最终确定任何建议前调用此函数。

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}

根据风险偏好进行适当性检查

让建议与用户声明的风险承受能力相匹配。不应向保守型投资者推荐高贝塔成长股；激进型投资者则可以承受更高的波动性。

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))

限制建议频率

过于频繁地产生交易建议会鼓励过度交易，从而增加成本和风险。请在针对同一资产的两次建议之间设置冷却期。

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'))

止损与回撤熔断机制

除了单项持仓限制外，还应实施投资组合层面的熔断机制：如果投资组合总价值在单日内损失超过 X%，智能体就应停止提出新的买入建议，直到人工检查情况。

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))

日内交易者（PDT）规则限制什么

PDT 规则是金融智能体在 US 市场运行时必须遵守的监管要求。理解这项规则有助于防止智能体无意中触发保证金账户限制。

在交互式课程中回答

风险与合规防护机制回顾

金融代理的防护机制包括：仓位规模限制、禁止交易工具列表、PDT 规则检查、按波动率调整仓位规模、根据风险状况进行适当性匹配、对每条建议进行审计日志记录，以及在所有输出中注入免责声明。

在确定任何交易建议前，始终运行所有防护机制。

免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 60
课程: 239

免费开始学习查看完整课程

常见问题解答

「风险与合规护栏」课时是免费的吗？

是的 — 「风险与合规护栏」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「风险与合规护栏」这节课中我会学到什么？

仓位规模限制、禁止交易工具检查和审计日志记录你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 3 节课，共 4 节。

「风险与合规护栏」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents