AI Agent股票异动风控机器人实战（支持美股+A股）---掌握这套方法论，独立搭建股票异动风控机器人

1┌─────────────────────────────────┐
2│         告警层（通知你）          │  钉钉/飞书/邮件/短信
3├─────────────────────────────────┤
4│         规则层（判断异动）         │  偏离值/振幅/换手率/量比
5├─────────────────────────────────┤
6│         计算层（实时处理）         │  滑动窗口 + 指标计算
7├─────────────────────────────────┤
8│         数据层（获取行情）         │  Tushare / AKShare / 交易所API
9└─────────────────────────────────┘
10

python1import akshare as ak
2import pandas as pd
3
4def get_realtime_quotes():
5    """获取实时行情，每3秒刷新一次"""
6    df = ak.stock_zh_a_spot_em()
7    df = df[['代码', '名称', '最新价', '涨跌幅', '成交量', '换手率', '振幅']]
8    df.columns = ['code', 'name', 'price', 'pct_change', 'volume', 'turnover', 'amplitude']
9    return df
10

python1def calc_deviation(df, window=3):
2    """计算N日涨跌幅偏离值"""
3    df['pct_change_1d'] = df['pct_change']
4    df['benchmark'] = df['pct_change_1d'].rolling(window).sum()  # 同期大盘涨幅
5    df['deviation'] = df['pct_change_1d'] - df['benchmark']
6    df['cum_deviation'] = df['deviation'].rolling(window).sum()
7    return df
8
9def check_abnormal(row):
10    """异动判定规则"""
11    rules = {
12        'deviation': abs(row['cum_deviation']) >= 20,     # 偏离值±20%
13        'amplitude': row['amplitude'] >= 15,               # 振幅≥15%
14        'turnover': row['turnover'] >= 20,                 # 换手率≥20%
15        'volume_ratio': row['volume'] / row['volume_ma_5'] >= 3  # 量比≥3
16    }
17    return any(rules.values()), rules
18

python1import requests
2
3def send_alert(stock, reason, value):
4    """飞书机器人推送"""
5    webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"
6    message = {
7        "msg_type": "interactive",
8        "card": {
9            "config": {"wide_screen_mode": True},
10            "elements": [{
11                "tag": "markdown",
12                "content": f"## 🚨 异动告警\n"
13                           f"**{stock['name']}({stock['code']})**\n"
14                           f"原因: {reason}\n"
15                           f"数值: {value}\n"
16                           f"时间: {datetime.now().strftime('%H:%M:%S')}"
17            }]
18        }
19    }
20    requests.post(webhook_url, json=message)
21

python1def backtest(df, start_date, end_date):
2    """用历史数据验证规则有效性"""
3    df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]
4    signals = []
5    for i in range(len(df)):
6        window = df.iloc[max(0,i-2):i+1]
7        if len(window) < 3: continue
8        abnormal, rules = check_abnormal(window.iloc[-1])
9        if abnormal:
10            signals.append({
11                'date': window.iloc[-1]['date'],
12                'code': window.iloc[-1]['code'],
13                'reason': [k for k,v in rules.items() if v][0]
14            })
15    return pd.DataFrame(signals)
16
17# 跑一遍2025年全年数据，看看误报率多少
18results = backtest(hist_df, '2025-01-01', '2025-12-31')
19print(f"全年触发{len(results)}次，误报率{calc_false_positive(results)}%")
20

dockerfile1FROM python:3.11-slim
2WORKDIR /app
3COPY requirements.txt .
4RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
5COPY . .
6CMD ["python", "main.py"]
7