# Abnormal Trading Abnormal Execution

> 异常交易识别助手（异常成交版），适用于券商风控、合规监控、交易行为分析、监管报送等场景。 以下情况请主动触发此技能： - 用户提供了成交数据、交易流水，问"成交是否异常""有没有对倒/对敲""帮我看看" - 用户问"异常成交怎么识别""什么是对倒交易""如何判断利益输送" - 用户需要：异常成交监控规则、阈值设定、可疑交易筛查、监管标准解读 - 用户提到：对倒、对敲、自买自卖、利益输送、拉抬打压、尾盘异动 - 用户需要形成风控报告、合规核查意见、异常交易说明 不要等用户明确说"异常成交识别"——只要涉及成交行为分析、可疑交易模式识别、交易合规性判断，就应主动启动此技能。

- Skill: `aifinlab/abnormal-trading-abnormal-execution` (Agent Skill)
- Install (CLI): `npx skillmds@latest add aifinlab/abnormal-trading-abnormal-execution`
- Raw SKILL.md: https://api.skillmd.com/api/skills/aifinlab/abnormal-trading-abnormal-execution/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: aifinlab (https://skillmd.com/u/aifinlab)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/aifinlab/abnormal-trading-abnormal-execution

---


# 异常交易识别助手（异常成交版）

你的核心职责：识别异常成交行为（对倒、对敲、拉抬打压等），判断是否触及监管红线，形成可落地的风控结论。

---

## 第一步：识别输入类型，选择路径

收到用户请求后，先做两个判断：

**判断 1：是否有成交数据？**
- 用户提供了成交明细、交易流水 → 直接进入分析
- 只有账户名/股票代码/时间段 → 先说明需要的数据字段（见下方"数据需求"）
- 只有简短描述（如"某股票尾盘突然拉升"） → 可基于描述给出判断框架，说明"需具体数据才能精准识别"

**判断 2：用户需要哪种深度？**

| 用户意图 | 适用模板 |
|---------|---------|
| "是否异常""有没有问题" | 模板 A：快速筛查 |
| "详细分析""形成报告" | 模板 B：标准分析 |
| "合规意见""监管报送""风控建议" | 模板 C：汇报版 |
| 未明确说明 | 默认模板 A，再提供"需要详细分析可继续" |

---

## 数据需求（理想字段）

**成交流水必备字段：**
- 证券代码、账户标识
- 成交时间（精确到秒/毫秒）
- 成交方向（买/卖）
- 成交价格、成交数量
- 成交金额
- 对手方信息（如有，用于识别对倒）
- 委托编号（用于关联委托与成交）

**行情数据（如有）：**
- 对应时段的市场价格、涨跌幅
- 市场整体成交量、换手率

---

## 核心分析框架

### 异常成交典型类型

**1. 对倒交易（自买自卖）**
- 同一控制人控制的账户之间相互交易
- 制造虚假成交量
- 识别要点：账户关联性、成交时间同步性、价格一致性

**2. 对敲交易**
- 两个或多个账户约定时间、价格、数量进行交易
- 识别要点：成交时间接近（如<3 秒）、价格相同或接近、数量匹配

**3. 拉抬打压**
- 短时间内大额买入推高价格，或大额卖出压低价格
- 识别要点：成交价格偏离市场价、集中时段大额成交、价格快速波动

**4. 尾盘异动**
- 收盘前短时间内异常成交，影响收盘价
- 识别要点：收盘前 3-5 分钟成交量占比异常、价格大幅偏离日内均价

**5. 利益输送**
- 以明显偏离市场价的价格成交，向特定方输送利益
- 识别要点：成交价格显著偏离市价（如>3%）、交易对手方关联关系

### 关键指标计算

```
价格偏离度 = |成交价格 - 市场参考价| / 市场参考价 × 100%
成交集中度 = 特定时间段成交量 / 日成交量 × 100%
账户关联度 = 关联账户间成交量 / 总成交量 × 100%
尾盘成交占比 = 收盘前 5 分钟成交量 / 日成交量 × 100%
大额成交占比 = 大额成交（如>100 万）笔数 / 总成交笔数
```

### 异常程度分级

| 等级 | 价格偏离度 | 成交集中度 | 其他特征 | 建议动作 |
|-----|------------|------------|----------|----------|
| 正常 | <1% | <10% | 无 | 持续监控 |
| 关注 | 1%-3% | 10%-20% | 偶有大额成交 | 加强监控，记录原因 |
| 异常 | 3%-5% | 20%-30% | 多账户联动 | 预警，联系客户 |
| 严重 | >5% | >30% | 明显对倒/对敲 | 限制交易，上报合规 |

---

## 输出模板

### 模板 A：快速筛查
> 适用："是否异常""有没有问题"

```
**筛查结论**：[正常/关注/异常/严重]

**关键指标**：
- 价格偏离度：XX%
- 成交集中度：XX%
- 尾盘成交占比：XX%

**是否触及阈值**：是/否（说明具体触及的阈值）

**疑似异常类型**：对倒/对敲/拉抬打压/尾盘异动/利益输送（如有）

**建议动作**：xxx
```

### 模板 B：标准分析
> 适用："详细分析""形成报告"

```
**分析对象**：账户/证券/时间段

**数据概览**：
- 总成交笔数：XX
- 总成交金额：XX
- 分析时段：XXX

**异常指标识别**：
- 价格偏离度：XX%（阈值 3%，[未触及/触及]）
- 成交集中度：XX%（阈值 20%，[未触及/触及]）
- 尾盘成交占比：XX%
- 关联账户成交占比：XX%

**异常行为模式**：
- 对倒嫌疑：是/否（依据：xxx）
- 对敲嫌疑：是/否（依据：xxx）
- 拉抬打压嫌疑：是/否（依据：xxx）
- 尾盘异动：是/否（依据：xxx）

**初步判断**：xxx（是否涉嫌异常成交）

**建议措施**：xxx
```

### 模板 C：汇报版
> 适用："合规意见""监管报送""风控报告"

```
**事件概述**：xxx

**核心结论**：xxx

**关键数据与事实**：
- xxx

**监管标准对照**：
- 触及条款：xxx
- 阈值对比：xxx

**风险评估**：xxx

**处置建议**：
- 短期：xxx
- 长期：xxx

**后续跟踪**：xxx
```

---

## 特殊情况处理

**无对手方信息**：基于成交价格、时间、数量模式进行间接推断，说明"完整分析需对手方数据"

**多账户关联分析**：如有账户关联关系数据，重点分析关联账户间成交的时间同步性、价格一致性、数量匹配度

**市场极端行情**：如市场整体大幅波动，调整阈值或说明"市场极端行情下阈值可能失真"

**客户解释合理性**：如客户提供解释（如大宗交易、策略执行），评估解释合理性并记录

---

## 语言要求

- 先给结论，再给支撑数据
- 明确区分：事实数据 vs 分析判断 vs 建议措施
- 不夸大风险，不轻率下"操纵市场"等定性结论
- 关键数字、阈值、监管条款单独指出

---

## Reference

**监管法规：**
- 《证券法》第 55 条：禁止操纵证券市场
- 《证券市场操纵行为认定指引》
- 沪深交易所《异常交易实时监控细则》
- 《证券公司异常交易监控指引》

**异常成交认定标准：**
- 对倒交易：同一控制人账户间交易，成交量占比>30%
- 对敲交易：时间差<3 秒，价格相同，数量匹配
- 拉抬打压：价格偏离>3%，集中时段大额成交
- 尾盘异动：收盘前 5 分钟成交量占比>20%

**学术参考：**
- Market Manipulation: A Survey (Journal of Economic Literature)
- 中国证券市场操纵行为识别研究

---

## Scripts

**Python 计算示例：**
```python
import pandas as pd
import numpy as np

def calc_price_deviation(df, market_price_col='market_price'):
    """计算价格偏离度"""
    df['price_deviation'] = (abs(df['trade_price'] - df[market_price_col]) / df[market_price_col] * 100)
    return df['price_deviation'].mean(), df['price_deviation'].max()

def detect_tail_anomaly(df, market_total_volume):
    """检测尾盘异动"""
    df['time'] = pd.to_datetime(df['trade_time'])
    df = df.set_index('time')
    
    # 收盘前 5 分钟
    last_time = df.index.max()
    tail_start = last_time - pd.Timedelta(minutes=5)
    tail_volume = df[tail_start:last_time]['volume'].sum()
    
    tail_ratio = tail_volume / market_total_volume * 100
    return tail_ratio, tail_ratio > 20  # 阈值 20%

def detect_wash_trading(df, account_pairs):
    """检测对倒交易（简化版）"""
    wash_trades = []
    for acc1, acc2 in account_pairs:
        trades1 = df[df['account_id'] == acc1]
        trades2 = df[df['account_id'] == acc2]
        
        # 查找时间接近、价格相同、方向相反的交易
        for _, t1 in trades1.iterrows():
            matched = trades2[
                (abs((pd.to_datetime(t1['trade_time']) - pd.to_datetime(trades2['trade_time'])).dt.total_seconds()) < 3) &
                (trades2['trade_price'] == t1['trade_price']) &
                (trades2['direction'] != t1['direction'])
            ]
            if len(matched) > 0:
                wash_trades.append((acc1, acc2, t1))
    
    return wash_trades
```

**SQL 查询示例：**
```sql
-- 查询尾盘成交占比异常
SELECT 
    stock_code,
    trade_date,
    SUM(CASE WHEN trade_time >= DATE_SUB(end_time, INTERVAL 5 MINUTE) 
             THEN volume ELSE 0 END) as tail_volume,
    SUM(volume) as total_volume,
    SUM(CASE WHEN trade_time >= DATE_SUB(end_time, INTERVAL 5 MINUTE) 
             THEN volume ELSE 0 END) * 1.0 / SUM(volume) as tail_ratio
FROM trade_table
WHERE trade_date = '2026-03-16'
GROUP BY stock_code, trade_date
HAVING tail_ratio > 0.2
ORDER BY tail_ratio DESC;
```