C#量化数据获取实战：从基础实现到生产环境优化

1次阅读

没有评论

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

最近在研究量化交易，发现数据获取这块坑特别多。API 动不动就限流，网络波动导致断连，还有各种数据格式不统一的问题。折腾了好几个版本后，终于总结出一套稳定可用的方案，今天把核心实现和踩坑经验分享给大家。

刚开始对接金融数据 API 时，经常遇到这些让人头疼的问题：

网络延迟波动 ：同一个 API 在不同时间段响应速度可能差 10 倍以上
严格限流政策 ：比如 Tushare Pro 的 500 次 / 分钟限制，一不小心就触发封禁
连接稳定性差 ：尤其是免费 API，经常出现 TCP 连接意外中断

有次做回测时，因为没处理好重试机制，导致获取 300 支股票历史数据时漏了 17 支，最终回测结果完全失真 …

实测了几种主流的 HTTP 客户端方案：

原始 HttpClient
优点：完全可控，性能最好
缺点：要自己处理连接池、重试等机制
WebSocket 长连接
适用场景：实时行情推送
坑点：心跳维护复杂，断线恢复成本高
Refit 声明式客户端
优点：接口定义简洁
缺点：灵活性较差，不适合复杂重试场景

最后选择用 HttpClient + Polly + TPL Dataflow 的组合，下面是具体实现。

使用 Polly 实现带指数退避的重试策略：

var retryPolicy = Policy
    .Handle<HttpRequestException>()
    .OrResult<HttpResponseMessage>(r => !r.IsSuccessStatusCode)
    .WaitAndRetryAsync(
        retryCount: 3,
        sleepDurationProvider: retryAttempt => 
            TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
        onRetry: (exception, delay, context) => 
            logger.Warn($"请求失败，{delay.TotalSeconds}s 后重试..."));

关键点：

对 429 状态码自动生效
首次重试间隔 2 秒，第二次 4 秒，避免雪崩
记录重试日志方便排查

用 TPL Dataflow 处理并发下载和解析：

var downloader = new TransformBlock<string, RawData>(async url =>
{using var semaphore = new SemaphoreSlim(10);
    await semaphore.WaitAsync();
    try {return await client.GetAsync(url);
    } finally {semaphore.Release();
    }
}, new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = 20});

var parser = new TransformBlock<RawData, ParsedData>(data => 
    JsonSerializer.Deserialize<ParsedData>(data));

downloader.LinkTo(parser);

采用两级缓存提升性能：

内存缓存：用 MemoryCache 缓存高频访问的元数据
本地持久化：将历史数据按日期分片存储为 Parquet 文件

public async Task<List<StockData>> FetchBatchDataAsync(IEnumerable<string> symbols)
{
    // 1. 认证处理
    if (token.IsExpired)
        await RefreshTokenAsync();

    // 2. 并发控制
    var throttler = new SemaphoreSlim(initialCount: 5);

    var tasks = symbols.Select(async symbol => {await throttler.WaitAsync();
        try {
            // 3. 带重试的执行
            return await retryPolicy.ExecuteAsync(() => FetchSingleDataAsync(symbol));
        } finally {throttler.Release();
        }
    });

    return (await Task.WhenAll(tasks))
        .Where(x => x != null)
        .ToList();}

// 在 Startup.cs 中
services.AddMetrics(registry =>
{
    registry.Measurement.Register(
        "api.latency", 
        () => new[] {lastLatency.TotalMilliseconds});
});

Policy.Handle<TimeoutException>()
    .CircuitBreaker(
        exceptionsAllowedBeforeBreaking: 3,
        durationOfBreak: TimeSpan.FromMinutes(1));

{
  "Timestamp": "2023-08-20T14:15:22Z",
  "Level": "Warning",
  "Message": "API 响应缓慢",
  "Properties": {
    "Endpoint": "/quotes",
    "Latency": 1250,
    "Symbol": "AAPL"
  }
}