跳转至

数据流合并

三个数据文件order.csv snap.csv trans.csv 分别代表股票的订单、快照以及成交数据。
这三组数据是由交易所发布的三个数据源的实时数据流。数据中的第一列time代表数据发布的时间，第二列price代表发布的价格。
我们现在需要根据数据发布的时间，将三种源数据流按照接近交易所实时发布时候我们从三个数据源收到的顺序去
合并成一个数据流。如果遇到相同发布时间，那么按照trans snap order这个顺序来确定先后。
合并成的这个数据流输出结果按照发布时间 发布价格 以及数据类型三列time,price,type。
type可以用o s t三个字母代表三种数据源数据。

假设order.csv数据是:
time,price
09:30:00.000,10.67
09:30:00.000,10.67

trans.csv数据:
time,price
09:30:00.000,0
09:30:00.030,0

snap.csv数据是：
time,price
09:30:00.000,11.81
09:30:03.000,11.82

那么输出result.csv是:
time,price,type
09:30:00.000,0,t
09:30:00.000,11.81,s
09:30:00.000,10.67,o
09:30:00.000,10.67,o
09:30:00.030,0,t
09:30:03.000,11.82,s

在linux环境下，实现c++版本的处理这个需求的代码。并且计时剔除数据读取和写入的时间。
附加思考:
目前这个是一只股票的数据流合并，实际中每个数据源中的股票有5000多只，而且数据源也不止这三种，
而且用户会随机从这些股票中选出一个子集合出来（比如选出300只股票），合并数据流。
因此性能是我们不得不考虑的问题，如何优化这种场景？

#include <algorithm>
#include <chrono>
#include <fstream>
#include <iostream>
#include <sstream>
#include <string>
#include <vector>

// 定义数据类型的优先级，trans(t) > snap(s) > order(o)
const int TYPE_PRIORITY[256] = {['t'] = 3, ['s'] = 2, ['o'] = 1};

// 数据结构表示一条记录
struct DataRecord {
  std::string time;
  double price;
  char type;

  // 比较运算符，用于排序
  bool operator<(const DataRecord& other) const {
    if (time != other.time) {
      return time < other.time;
    }
    // 时间相同时，按照优先级排序
    return TYPE_PRIORITY[type] < TYPE_PRIORITY[other.type];
  }
};

// 从CSV文件读取数据
void readCSV(const std::string& filename, char type,
             std::vector<DataRecord>& data) {
  std::ifstream file(filename);
  if (!file.is_open()) {
    std::cerr << "无法打开文件: " << filename << std::endl;
    return;
  }

  std::string line;
  // 跳过表头
  std::getline(file, line);

  while (std::getline(file, line)) {
    std::istringstream iss(line);
    std::string timeStr, priceStr;

    // 解析CSV行
    if (std::getline(iss, timeStr, ',') && std::getline(iss, priceStr, ',')) {
      DataRecord record;
      record.time = timeStr;
      record.price = std::stod(priceStr);
      record.type = type;
      data.push_back(record);
    }
  }
}

int main() {
  std::vector<DataRecord> allData;

  readCSV("order.csv", 'o', allData);
  readCSV("snap.csv", 's', allData);
  readCSV("trans.csv", 't', allData);

  auto start = std::chrono::high_resolution_clock::now();

  std::sort(allData.begin(), allData.end());

  auto end = std::chrono::high_resolution_clock::now();
  std::chrono::duration<double, std::micro> elapsed = end - start;
  std::cerr << "数据处理时间: " << elapsed.count() << " 微秒" << std::endl;

  std::ofstream outFile("result.csv");
  if (!outFile.is_open()) {
    std::cerr << "无法创建输出文件" << std::endl;
    return 1;
  }

  // 写入表头
  outFile << "time,price,type" << std::endl;

  // 写入数据
  for (const auto& record : allData) {
    outFile << record.time << "," << record.price << "," << record.type
            << std::endl;
  }

  return 0;
}

// 对于包含 5000 多只股票和更多数据源的场景，可以考虑以下优化：
// 并行处理：对不同股票进行并行处理，使用多线程读取和处理不同的数据源
// 内存优化：使用内存映射文件处理大型
// CSV，采用更紧凑的数据结构，如用整数表示时间而非字符串
// 预排序与归并：每个数据源可能已经按时间排序，使用多路归并算法而非全量排序
// 数据过滤：在读取阶段就过滤掉不需要的股票数据，减少处理量，可以使用哈希表快速查找用户选择的股票子集
// 数据格式：考虑使用二进制格式替代 CSV，减少解析开销