数据导入

S# 实现了一个从 CSV 文件导入市场数据的子系统。主要类位于 StockSharp.Algo.Import 命名空间中。

CsvParser — 基础解析器

CsvParser 类执行 CSV 文件解析并将行转换为 S# 消息。

构造函数: (DataType dataType, IEnumerable<FieldMapping> fields)
ColumnSeparator — 列分隔符（默认 ","）。
LineSeparator — 行分隔符（默认 CRLF）。
SkipFromHeader — 从文件开头跳过的行数（默认 0）。
IgnoreNonIdSecurities — 忽略具有无法识别的交易品种的行（默认 true）。
Parse(Stream) — 解析方法，返回 IAsyncEnumerable<Message>。

var fields = FieldMappingRegistry.CreateFields(DataType.Ticks);
var parser = new CsvParser(DataType.Ticks, fields)
{
    ColumnSeparator = ";",
    SkipFromHeader = 1
};

await using var stream = File.OpenRead("trades.csv");

await foreach (var msg in parser.Parse(stream))
{
    // 处理每条消息
}

CsvImporter — 带存储节省的导入

CsvImporter 类继承自 CsvParser，增加了自动将数据保存到市场数据存储的功能。

构造函数: (DataType dataType, IEnumerable<FieldMapping> fields, ISecurityStorage securityStorage, IExchangeInfoProvider exchangeInfoProvider, Func<SecurityId, IMarketDataStorage> getStorage)
Import(Stream, Action<int> progress, CancellationToken) — 执行导入并返回 ValueTask<(int count, DateTime? lastTime)>。
UpdateDuplicateSecurities — 是否更新重复的交易品种（默认 false）。
SecurityUpdated — 当交易品种更新时触发的事件。

var fields = FieldMappingRegistry.CreateFields(DataType.Ticks);

var importer = new CsvImporter(
    DataType.Ticks,
    fields,
    securityStorage,
    exchangeInfoProvider,
    secId => storageRegistry.GetStorage(secId, DataType.Ticks))
{
    ColumnSeparator = ";",
    SkipFromHeader = 1
};

await using var stream = File.OpenRead("trades.csv");
var (count, lastTime) = await importer.Import(
    stream,
    p => Console.WriteLine($"进度: {p}%"),
    token);

Console.WriteLine($"已导入 {count} 条记录，最后一条: {lastTime}");

字段映射 — 字段描述

FieldMapping 类描述了 CSV 文件列与消息属性之间的映射。

主要属性：

Name — 消息中的字段名称。
DisplayName — 显示名称。
Type — 值类型。
Order — 文件中的列索引（从 0 开始）。
IsRequired — 该字段是否为必填项。
Format — 解析格式 (例如，日期格式)。
DefaultValue — 默认值。
ZeroAsNull — 是否将零值解释为 null。

对于自定义值转换，请使用 FieldMappingValue。例如，您可以定义文本值到枚举的映射：

var sideField = fields.First(f => f.Name == "Side");
sideField.Order = 3;
sideField.Values.Add(new FieldMappingValue
{
    ValueFrom = "B",
    ValueTo = Sides.Buy
});
sideField.Values.Add(new FieldMappingValue
{
    ValueFrom = "S",
    ValueTo = Sides.Sell
});

FieldMappingRegistry — 标准字段注册表

静态类 FieldMappingRegistry 提供了一种创建标准字段集的方法：

CreateFields(DataType) — 返回指定数据类型的 FieldMapping 列表。

支持的数据类型：逐笔成交、K线、订单簿、Level1、订单日志、交易、交易品种、新闻和持仓。

导入设置 — 导入设置

ImportSettings 类将所有导入参数组合到一个配置对象中：

DataType — 被导入的数据类型。
FileName — 文件路径。
Directory — 用于文件搜索的目录。
FileMask — 文件搜索掩码 (例如， *.csv)。
ColumnSeparator — 列分隔符。
SkipFromHeader — 要跳过的行数。
SelectedFields — 选择导入的字段。
UpdateDuplicateSecurities — 是否更新重复的交易品种。

辅助方法：

GetFiles(IFileSystem) — 获取与掩码匹配的文件列表。
FillParser(CsvParser) — 填充解析器设置。
FillImporter(CsvImporter) — 填充导入器设置。

var settings = new ImportSettings
{
    DataType = DataType.Ticks,
    FileName = "trades.csv",
    ColumnSeparator = ";",
    SkipFromHeader = 1
};

var fields = FieldMappingRegistry.CreateFields(settings.DataType);
// 配置列顺序
fields[0].Order = 0; // SecurityId
fields[1].Order = 1; // Date
fields[2].Order = 2; // Price
fields[3].Order = 3; // Volume

var importer = new CsvImporter(
    settings.DataType,
    fields,
    securityStorage,
    exchangeInfoProvider,
    secId => storageRegistry.GetStorage(secId, settings.DataType));

settings.FillImporter(importer);

await using var stream = File.OpenRead(settings.FileName);
var (count, lastTime) = await importer.Import(stream, p => { }, token);

另请参阅

数据导出

数据存储