利用AKShare技能构建高效金融数据采集系统：实战避坑指南

1次阅读

共计 2555 个字符，预计需要花费 7 分钟才能阅读完成。

金融数据采集过程中常遇到接口限流导致获取失败、数据字段频繁变更引发解析错误、不同数据源格式差异大需要额外清洗等问题。这些问题不仅增加了开发成本，还影响了数据获取的效率和稳定性。

特性	AKShare	Tushare
数据覆盖范围	股票、基金、期货、期权等	主要侧重股票市场数据
接口稳定性	较高	一般
数据更新频率	实时 / 日级	日级
免费额度	完全免费	部分接口需要付费
社区支持	活跃	一般

import akshare as ak
import pandas as pd
import logging
from retrying import retry
from functools import lru_cache

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='financial_data.log'
)
logger = logging.getLogger(__name__)

# 带重试机制的请求封装
@retry(stop_max_attempt_number=3, wait_fixed=2000)
def fetch_with_retry(func, **kwargs):
    try:
        data = func(**kwargs)
        if data.empty:
            raise ValueError('Empty DataFrame returned')
        return data
    except Exception as e:
        logger.error(f'Fetch data failed: {str(e)}')
        raise

# 数据字段校验
REQUIRED_STOCK_COLUMNS = ['代码', '名称', '最新价', '涨跌幅']

def validate_columns(df, required_columns):
    missing = set(required_columns) - set(df.columns)
    if missing:
        raise ValueError(f'Missing required columns: {missing}')
    return True

# 示例使用
if __name__ == '__main__':
    try:
        stock_data = fetch_with_retry(
            ak.stock_zh_a_spot,
            symbol='sh600000'
        )
        validate_columns(stock_data, REQUIRED_STOCK_COLUMNS)
        print(stock_data.head())
    except Exception as e:
        logger.error(f'Main process failed: {str(e)}')

异步请求实现

import asyncio
import aiohttp

async def fetch_async(session, url):
    async with session.get(url) as response:
        return await response.json()

async def fetch_multiple_stocks(symbols):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for symbol in symbols:
            url = f'https://api.example.com/stocks/{symbol}'
            tasks.append(fetch_async(session, url))
        return await asyncio.gather(*tasks)

本地缓存策略

import sqlite3
from datetime import datetime

# SQLite 缓存实现
class DataCache:
    def __init__(self, db_path='financial_data.db'):
        self.conn = sqlite3.connect(db_path)
        self._create_tables()

    def _create_tables(self):
        cursor = self.conn.cursor()
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS stock_data (
            symbol TEXT PRIMARY KEY,
            data TEXT,
            updated_at TIMESTAMP
        )
        ''')
        self.conn.commit()

    @lru_cache(maxsize=128)
    def get_cached_data(self, symbol):
        cursor = self.conn.cursor()
        cursor.execute(
            'SELECT data FROM stock_data WHERE symbol=?',
            (symbol,)
        )
        result = cursor.fetchone()
        return result[0] if result else None

    def update_cache(self, symbol, data):
        cursor = self.conn.cursor()
        cursor.execute('INSERT OR REPLACE INTO stock_data VALUES (?, ?, ?)',
            (symbol, data, datetime.now())
        )
        self.conn.commit()
        self.get_cached_data.cache_clear()