交易员持仓报告

交易员持仓报告 (COT) 由 CFTC 每周五发布。CFTC COT 报告提供了每周二期货和期货期权市场未平仓合约的详细分类，涵盖了 20 个或更多交易员持有等于或超过 CFTC 设定报告水平的头寸的市场。

监管机构模块

obb.regulators 模块包含行业监管机构和机构发布的数据。这些数据不特定于任何特定资产类别，信息向公众开放。COT 报告有两个端点：

• obb.regulators.cftc.cot()
• obb.regulators.cftc.cot_search()

openbb-cftc 扩展（包含在 pip install openbb 中）提供对完整历史报告的程序化访问，可追溯到 1995 年。

cot_search()

COT 搜索

obb.regulators.cftc.cot_search() 端点是当前报告的精选列表。该列表可以通过部分匹配进行搜索 - 例如，"resources" - 它们按类别和子类别分类。使用空查询获取完整列表。

指数 - 标普 500、纳斯达克 100、道琼斯工业平均指数、罗素 1000 和 2000、VIX、彭博商品指数等 - 可以使用关键词 "index" 找到。

下面的示例显示了所有标普相关报告。

# 搜索标普相关的 COT 报告
reports = obb.regulators.cftc.cot_search("s&p").to_dataframe()
print("标普相关的 COT 报告:")
print(reports)

代码	名称	商品名称	类别	子类别
43874A	S&P 500 年度股息指数	股息指数	金融工具	股票指数
43874Q	S&P 500 季度股息指数	股息指数	金融工具	股票指数
13874I	E-迷你标普科技指数	标普广基股票指数	金融工具	股票指数
138748	E-迷你标普消费必需品指数	标普广基股票指数	金融工具	股票指数
13874A	E-迷你标普 500	标普广基股票指数	金融工具	股票指数
138749	E-迷你标普能源指数	标普广基股票指数	金融工具	股票指数

cot()

概述

每周报告以平面时间序列形式返回。id 参数接受灵活输入，使用 "code" 提供精确匹配，返回完整历史。有关输入符号的更多信息，请参阅下面的参数部分。

# 获取标普 500 的 COT 数据
sp = obb.regulators.cftc.cot("13874+")
df_sp = sp.to_dataframe()

# 显示最近 5 个报告期的未平仓合约变化
print("标普 500 COT 数据:")
print(df_sp.tail(5))

基本 COT 数据获取

搜索可用报告

# 搜索所有可用的 COT 报告
all_reports = obb.regulators.cftc.cot_search("").to_dataframe()
print(f"总共有 {len(all_reports)} 个 COT 报告")

# 按类别分组
category_counts = all_reports['category'].value_counts()
print("\n按类别分布:")
print(category_counts)

# 搜索特定商品
gold_reports = obb.regulators.cftc.cot_search("gold").to_dataframe()
print(f"\n黄金相关报告 ({len(gold_reports)} 个):")
print(gold_reports[['code', 'name', 'commodity_name']])

获取具体商品的 COT 数据

def get_cot_data(commodity_code, commodity_name=""):
    """
    获取特定商品的 COT 数据
    """
    try:
        cot_data = obb.regulators.cftc.cot(commodity_code)
        df = cot_data.to_dataframe()
        
        if df.empty:
            print(f"未找到 {commodity_name} ({commodity_code}) 的 COT 数据")
            return None
        
        print(f"{commodity_name} COT 数据:")
        print(f"  数据范围: {df['date'].min()} 到 {df['date'].max()}")
        print(f"  总记录数: {len(df)}")
        
        # 显示最新数据
        latest = df.iloc[-1]
        print(f"  最新报告日期: {latest['date']}")
        if 'open_interest' in df.columns:
            print(f"  未平仓合约: {latest['open_interest']:,}")
        
        return df
        
    except Exception as e:
        print(f"获取 {commodity_name} COT 数据失败: {e}")
        return None

# 获取主要商品的 COT 数据
commodities = [
    ("001602", "玉米"),
    ("002602", "大豆"),
    ("067651", "原油"),
    ("088691", "黄金"),
    ("13874A", "标普 500")
]

cot_data = {}
for code, name in commodities:
    data = get_cot_data(code, name)
    if data is not None:
        cot_data[name] = data

COT 数据分析

持仓结构分析

def analyze_cot_positions(df, commodity_name):
    """
    分析 COT 持仓结构
    """
    if df.empty:
        return None
    
    # 获取最新数据
    latest = df.iloc[-1]
    
    analysis = {
        'commodity': commodity_name,
        'report_date': latest['date'],
        'total_open_interest': latest.get('open_interest', 0)
    }
    
    # 分析不同类型交易员的持仓
    trader_types = [
        ('commercial_long', 'commercial_short', '商业交易员'),
        ('non_commercial_long', 'non_commercial_short', '非商业交易员'),
        ('non_reportable_long', 'non_reportable_short', '非报告交易员')
    ]
    
    for long_col, short_col, trader_type in trader_types:
        if long_col in df.columns and short_col in df.columns:
            long_pos = latest.get(long_col, 0)
            short_pos = latest.get(short_col, 0)
            net_pos = long_pos - short_pos
            
            analysis[f'{trader_type}_多头'] = long_pos
            analysis[f'{trader_type}_空头'] = short_pos
            analysis[f'{trader_type}_净持仓'] = net_pos
    
    return analysis

# 分析主要商品的持仓结构
for name, data in cot_data.items():
    if data is not None:
        analysis = analyze_cot_positions(data, name)
        if analysis:
            print(f"\n{name} 持仓结构分析:")
            print(f"  报告日期: {analysis['report_date']}")
            print(f"  总未平仓合约: {analysis['total_open_interest']:,}")
            
            # 显示净持仓
            for key, value in analysis.items():
                if '净持仓' in key:
                    direction = "看多" if value > 0 else "看空" if value < 0 else "中性"
                    print(f"  {key}: {value:,} ({direction})")

持仓变化趋势分析

def analyze_cot_trends(df, periods=4):
    """
    分析 COT 持仓变化趋势
    """
    if len(df) < periods:
        print("数据不足，无法分析趋势")
        return None
    
    # 获取最近几期数据
    recent_data = df.tail(periods).copy()
    
    # 计算变化
    trends = {}
    
    # 分析未平仓合约变化
    if 'open_interest' in recent_data.columns:
        oi_change = recent_data['open_interest'].iloc[-1] - recent_data['open_interest'].iloc[0]
        oi_pct_change = (oi_change / recent_data['open_interest'].iloc[0]) * 100
        trends['未平仓合约变化'] = {
            'absolute': oi_change,
            'percentage': oi_pct_change
        }
    
    # 分析净持仓变化
    position_columns = [
        ('commercial_long', 'commercial_short', '商业交易员'),
        ('non_commercial_long', 'non_commercial_short', '非商业交易员')
    ]
    
    for long_col, short_col, trader_type in position_columns:
        if long_col in recent_data.columns and short_col in recent_data.columns:
            # 计算净持仓
            recent_data[f'{trader_type}_net'] = recent_data[long_col] - recent_data[short_col]
            
            # 计算变化
            net_change = recent_data[f'{trader_type}_net'].iloc[-1] - recent_data[f'{trader_type}_net'].iloc[0]
            trends[f'{trader_type}净持仓变化'] = net_change
    
    return trends

# 分析趋势
for name, data in cot_data.items():
    if data is not None:
        trends = analyze_cot_trends(data, periods=8)  # 分析最近8周
        if trends:
            print(f"\n{name} 持仓趋势分析 (最近8周):")
            for trend_name, change in trends.items():
                if isinstance(change, dict):
                    print(f"  {trend_name}: {change['absolute']:,} ({change['percentage']:+.2f}%)")
                else:
                    direction = "增加" if change > 0 else "减少" if change < 0 else "无变化"
                    print(f"  {trend_name}: {change:,} ({direction})")

市场情绪指标

def calculate_cot_sentiment(df):
    """
    计算基于 COT 数据的市场情绪指标
    """
    if df.empty:
        return None
    
    latest = df.iloc[-1]
    
    # 计算非商业交易员净持仓比例（投机情绪指标）
    if all(col in df.columns for col in ['non_commercial_long', 'non_commercial_short', 'open_interest']):
        nc_long = latest['non_commercial_long']
        nc_short = latest['non_commercial_short']
        total_oi = latest['open_interest']
        
        nc_net = nc_long - nc_short
        nc_net_ratio = (nc_net / total_oi) * 100 if total_oi > 0 else 0
        
        # 情绪判断
        if nc_net_ratio > 10:
            sentiment = "极度看多"
        elif nc_net_ratio > 5:
            sentiment = "看多"
        elif nc_net_ratio > -5:
            sentiment = "中性"
        elif nc_net_ratio > -10:
            sentiment = "看空"
        else:
            sentiment = "极度看空"
        
        return {
            'net_ratio': nc_net_ratio,
            'sentiment': sentiment,
            'net_position': nc_net
        }
    
    return None

# 计算市场情绪
print("\n市场情绪指标 (基于非商业交易员持仓):")
for name, data in cot_data.items():
    if data is not None:
        sentiment = calculate_cot_sentiment(data)
        if sentiment:
            print(f"  {name}: {sentiment['sentiment']} (净持仓比例: {sentiment['net_ratio']:+.2f}%)")

COT 数据可视化准备

def prepare_cot_chart_data(df, commodity_name, periods=52):
    """
    准备 COT 数据用于图表展示
    """
    if len(df) < periods:
        chart_data = df.copy()
    else:
        chart_data = df.tail(periods).copy()
    
    # 计算净持仓
    if all(col in chart_data.columns for col in ['commercial_long', 'commercial_short']):
        chart_data['commercial_net'] = chart_data['commercial_long'] - chart_data['commercial_short']
    
    if all(col in chart_data.columns for col in ['non_commercial_long', 'non_commercial_short']):
        chart_data['non_commercial_net'] = chart_data['non_commercial_long'] - chart_data['non_commercial_short']
    
    # 选择关键列
    key_columns = ['date', 'open_interest']
    if 'commercial_net' in chart_data.columns:
        key_columns.append('commercial_net')
    if 'non_commercial_net' in chart_data.columns:
        key_columns.append('non_commercial_net')
    
    chart_ready = chart_data[key_columns].copy()
    chart_ready['commodity'] = commodity_name
    
    return chart_ready

# 准备图表数据
chart_datasets = {}
for name, data in cot_data.items():
    if data is not None:
        chart_data = prepare_cot_chart_data(data, name, periods=26)  # 最近26周
        chart_datasets[name] = chart_data
        print(f"{name} 图表数据准备完成: {len(chart_data)} 个数据点")

多商品 COT 比较

def compare_cot_sentiment(commodities_data):
    """
    比较多个商品的 COT 情绪
    """
    comparison = []
    
    for name, data in commodities_data.items():
        if data is not None:
            sentiment = calculate_cot_sentiment(data)
            if sentiment:
                comparison.append({
                    'commodity': name,
                    'sentiment': sentiment['sentiment'],
                    'net_ratio': sentiment['net_ratio'],
                    'net_position': sentiment['net_position']
                })
    
    # 按净持仓比例排序
    comparison.sort(key=lambda x: x['net_ratio'], reverse=True)
    
    return comparison

# 比较分析
comparison_results = compare_cot_sentiment(cot_data)

print("\n商品 COT 情绪排名 (按非商业交易员净持仓比例):")
print("-" * 60)
for i, result in enumerate(comparison_results, 1):
    print(f"{i}. {result['commodity']}: {result['sentiment']} ({result['net_ratio']:+.2f}%)")

COT 数据监控系统

class COTMonitor:
    """COT 数据监控系统"""
    
    def __init__(self, watch_list=None):
        self.watch_list = watch_list or []
        self.historical_data = {}
    
    def add_commodity(self, code, name):
        """添加监控商品"""
        self.watch_list.append({'code': code, 'name': name})
        print(f"已添加 {name} ({code}) 到监控列表")
    
    def update_data(self):
        """更新所有监控商品的数据"""
        print("更新 COT 数据...")
        
        for commodity in self.watch_list:
            try:
                data = obb.regulators.cftc.cot(commodity['code'])
                df = data.to_dataframe()
                
                if not df.empty:
                    self.historical_data[commodity['name']] = df
                    print(f"✓ {commodity['name']}: {len(df)} 条记录")
                else:
                    print(f"✗ {commodity['name']}: 无数据")
                    
            except Exception as e:
                print(f"✗ {commodity['name']}: 更新失败 - {e}")
    
    def generate_report(self):
        """生成监控报告"""
        if not self.historical_data:
            print("无数据可报告，请先更新数据")
            return
        
        print("\nCOT 监控报告")
        print("=" * 50)
        
        for name, data in self.historical_data.items():
            print(f"\n{name}:")
            
            # 基本信息
            latest = data.iloc[-1]
            print(f"  最新报告: {latest['date']}")
            
            if 'open_interest' in data.columns:
                print(f"  未平仓合约: {latest['open_interest']:,}")
            
            # 情绪分析
            sentiment = calculate_cot_sentiment(data)
            if sentiment:
                print(f"  市场情绪: {sentiment['sentiment']}")
                print(f"  净持仓比例: {sentiment['net_ratio']:+.2f}%")
            
            # 趋势分析
            trends = analyze_cot_trends(data, periods=4)
            if trends and '非商业交易员净持仓变化' in trends:
                change = trends['非商业交易员净持仓变化']
                direction = "增加" if change > 0 else "减少" if change < 0 else "无变化"
                print(f"  4周趋势: {direction} ({change:,})")

# 使用示例
monitor = COTMonitor()

# 添加监控商品
monitor.add_commodity("001602", "玉米")
monitor.add_commodity("002602", "大豆")
monitor.add_commodity("067651", "原油")
monitor.add_commodity("088691", "黄金")
monitor.add_commodity("13874A", "标普500")

# 更新数据并生成报告
monitor.update_data()
monitor.generate_report()

最佳实践

1. 数据时效性：COT 报告每周五发布，数据截至周二
2. 持仓分类理解：区分商业交易员（套期保值者）和非商业交易员（投机者）
3. 趋势分析：关注持仓变化趋势而非绝对数值
4. 多维度分析：结合价格走势和持仓数据进行分析
5. 风险管理：COT 数据是参考指标，不应作为唯一决策依据

# 完整的 COT 分析工作流
def comprehensive_cot_analysis(commodity_code, commodity_name):
    """
    综合 COT 分析工作流
    """
    print(f"开始 {commodity_name} COT 综合分析...")
    print("=" * 50)
    
    # 1. 获取数据
    try:
        cot_data = obb.regulators.cftc.cot(commodity_code)
        df = cot_data.to_dataframe()
        
        if df.empty:
            print("未找到数据")
            return
        
        print(f"✓ 数据获取成功: {len(df)} 条记录")
        
    except Exception as e:
        print(f"✗ 数据获取失败: {e}")
        return
    
    # 2. 基本信息
    latest = df.iloc[-1]
    print(f"\n基本信息:")
    print(f"  最新报告日期: {latest['date']}")
    print(f"  数据历史: {df['date'].min()} 到 {df['date'].max()}")
    
    # 3. 持仓结构分析
    print(f"\n持仓结构分析:")
    analysis = analyze_cot_positions(df, commodity_name)
    if analysis:
        for key, value in analysis.items():
            if '净持仓' in key and isinstance(value, (int, float)):
                direction = "看多" if value > 0 else "看空" if value < 0 else "中性"
                print(f"  {key}: {value:,} ({direction})")
    
    # 4. 市场情绪
    print(f"\n市场情绪:")
    sentiment = calculate_cot_sentiment(df)
    if sentiment:
        print(f"  投机情绪: {sentiment['sentiment']}")
        print(f"  净持仓比例: {sentiment['net_ratio']:+.2f}%")
    
    # 5. 趋势分析
    print(f"\n趋势分析 (最近8周):")
    trends = analyze_cot_trends(df, periods=8)
    if trends:
        for trend_name, change in trends.items():
            if isinstance(change, dict):
                print(f"  {trend_name}: {change['percentage']:+.2f}%")
            elif isinstance(change, (int, float)):
                direction = "上升" if change > 0 else "下降" if change < 0 else "平稳"
                print(f"  {trend_name}: {direction}")
    
    print(f"\n{commodity_name} COT 分析完成!")
    return df

# 使用示例
gold_analysis = comprehensive_cot_analysis("088691", "黄金")