current-akshare-list

获取A股全量股票基础信息列表，并用实时行情接口交叉验证，输出「在交易」和「不在交易」两份清单。

环境要求

• Python 3.11+
• akshare SDK：pip install akshare
• 若接口封禁，需开启VPN或建立定时任务等待

工作流程

第一步：下载股票基础信息（可选，若已有可跳过）

从六个市场来源获取股票列表，合并去重：

import akshare as ak
import pandas as pd

sources = [
    ("沪A", lambda: ak.stock_info_sh_name_code(symbol="主板A股")),
    ("沪B", lambda: ak.stock_info_sh_name_code(symbol="主板B股")),
    ("沪创", lambda: ak.stock_zh_index_cons_ci(symbol="000688")),
    ("深A", lambda: ak.stock_info_sz_name_code(symbol="A股列表")),
    ("深B", lambda: ak.stock_info_sz_name_code(symbol="B股列表")),
    ("京A", lambda: ak.stock_info_bj_a_code_name()),
]

results = []
for name, fn in sources:
    try:
        df = fn()
        df["source"] = name
        results.append(df)
    except Exception as e:
        print(f"{name} 获取失败: {e}")

stocklist = pd.concat(results).drop_duplicates(subset=["code"])
stocklist.columns = ["Symbol", "ShortName", "ListedDate", "Source"]
stocklist.to_csv("stocklist.csv", index=False)

第二步：实时行情接口验证（核心）

依次尝试三个接口，遇到封堵则处理：

import akshare as ak
import json

def fetch_spot_data():
    """获取三接口实时行情数据"""
    data = {}
    
    # 1. A股实时行情
    try:
        df = ak.stock_zh_a_spot_em()
        codes = df[['代码','名称']].to_dict('records')
        with open('/tmp/akshare_a_codes.json', 'w') as f:
            json.dump(codes, f, ensure_ascii=False)
        data['a'] = len(codes)
        print(f"✅ A股成功: {len(codes)} 条")
    except Exception as e:
        print(f"❌ A股失败: {type(e).__name__}: {e}")
        raise ConnectionError(f"A股接口封禁: {e}")
    
    # 2. B股实时行情
    try:
        df = ak.stock_zh_b_spot_em()
        codes = df[['代码','名称']].to_dict('records')
        with open('/tmp/akshare_b_codes.json', 'w') as f:
            json.dump(codes, f, ensure_ascii=False)
        data['b'] = len(codes)
        print(f"✅ B股成功: {len(codes)} 条")
    except Exception as e:
        print(f"❌ B股失败: {e}")
    
    # 3. 科创板实时行情
    try:
        df = ak.stock_zh_kcb_spot()
        codes = df[['代码','名称']].to_dict('records')
        with open('/tmp/akshare_kcb_codes.json', 'w') as f:
            json.dump(codes, f, ensure_ascii=False)
        data['kcb'] = len(codes)
        print(f"✅ 科创板成功: {len(codes)} 条")
    except Exception as e:
        print(f"❌ 科创板失败: {e}")
    
    return data

# 执行获取
try:
    data = fetch_spot_data()
    print(f"完成: {data}")
except ConnectionError as e:
    print(f"接口封禁，需要处理...")
    raise

第三步：对比分析

import json, csv

# 读取三个接口数据
with open('/tmp/akshare_a_codes.json', 'r') as f:
    a_codes = set(item['代码'] for item in json.load(f))
with open('/tmp/akshare_b_codes.json', 'r') as f:
    b_codes = set(item['代码'] for item in json.load(f))
with open('/tmp/akshare_kcb_codes.json', 'r') as f:
    kcb_codes = set(item['代码'] for item in json.load(f))

# 读取 stocklist
stocklist_codes = set()
stocklist_info = {}
with open('stocklist.csv', 'r', encoding='utf-8-sig') as f:
    reader = csv.DictReader(f)
    for row in reader:
        code = row['Symbol'].strip()
        stocklist_codes.add(code)
        stocklist_info[code] = {'name': row.get('ShortName', ''), 'date': row.get('ListedDate', '')}

# 三接口合并（去重）
all_spot_codes = a_codes | b_codes | kcb_codes

# 对比分析
in_spot_not_stocklist = all_spot_codes - stocklist_codes  # 需补充
in_stocklist_not_spot = stocklist_codes - all_spot_codes  # notspot

print(f"需补充: {len(in_spot_not_stocklist)}")
print(f"notspot: {len(in_stocklist_not_spot)}")

第四步：接口封禁处理

当东财接口被封（ConnectionError、RemoteDisconnected）时：

方案A：用户手动处理

告知用户："东方财富接口被封，请开启VPN后重新执行，或选择建立定时任务自动重试"

方案B：自动建立定时任务

1. 创建 cron 任务，每30分钟重试东财接口
2. 若成功：执行全量对比 → 输出结果 → 删除自身任务
3. 若失败：输出"仍被封，等待下次尝试" → 保留任务继续重试
4. 最大重试次数：24次（12小时），超时后通知用户

# 定时任务 payload 示例
{
    "kind": "agentTurn",
    "message": "重试东财接口，若成功则完成全量对比并输出结果到指定路径",
    "sessionTarget": "isolated"
}

第五步：输出结果文件

VPN 环境说明

东方财富接口（stock_zh_a_spot_em）在大陆IP下经常被封：

• 开启VPN后：通常可稳定获取数据
• 无VPN：接口返回 ConnectionError 或 RemoteDisconnected

若遇封禁，建议：

1. 开启VPN后重试
2. 建立定时任务自动重试（适合夜间执行）

已知限制

• B股不在部分实时行情覆盖范围内，预期出现在 notspot 中
• 科创板全量通过指数成分股接口获取（非直接全量列表）