arrow(c++)改写empyrical系列1---用arrow读取基金净值数据并计算夏普率_arrow_assign_or_raise

用arrow c++版本读取了csv中的基金净值数据，然后计算了夏普率，比较尴尬的是，arrow c++版本计算耗费的时间却比python的empyrical版本耗费时间多。。。

arrow新手上路，第一次自己去实现功能，实现的大概率并不是最高效的方式，但是我也踏出了用arrow c++改写backtrader的第一步。

• 用arrow改写empyrical，就当练手了，目标是做成两个文件：empyrical.h和empyrical.cpp
• 用arrow和qt改写pyfolio, 实现更美观的界面，做成两个文件：pyfolio.h 和pyfolio.cpp
• 改写backtrader

c++版本的文件：

my_example.cc

#include <arrow/api.h>
#include <arrow/io/api.h>
#include "arrow/csv/api.h"
#include <arrow/compute/api.h>
#include <iostream>
#include <chrono>
//#include "../empyrical/empyrical.h"

arrow::Status RunMain(){
    auto start_time = std::chrono::high_resolution_clock::now();
    // 首先，我们需要设置一个可读文件对象，它允许我们将读取器指向磁盘上的正确数据。我们将重复使用这个对象，并将其重新绑定到多个文件中。
    std::shared_ptr<arrow::io::ReadableFile> infile;
    // 绑定输入文件到 "test_in.csv"
    ARROW_ASSIGN_OR_RAISE(infile, arrow::io::ReadableFile::Open("/home/yun/Documents/fund_nav.csv"));
    // (文档部分：CSV 表格声明)
    std::shared_ptr<arrow::Table> csv_table;
    // CSV 读取器有多个对象，用于不同选项。现在，我们将使用默认值。
    ARROW_ASSIGN_OR_RAISE(
        auto csv_reader,
        arrow::csv::TableReader::Make(
            arrow::io::default_io_context(), infile, arrow::csv::ReadOptions::Defaults(),
            arrow::csv::ParseOptions::Defaults(), arrow::csv::ConvertOptions::Defaults()));
    // 读取表格。
    ARROW_ASSIGN_OR_RAISE(csv_table, csv_reader->Read());

    // 输出显示Table的元数据信息
    // std::cout << "Table Metadata:" << std::endl;
    // std::cout << "Number of columns: " << csv_table->num_columns() << std::endl;
    // std::cout << "Number of rows: " << csv_table->num_rows() << std::endl;
    // std::cout << "Schema: " << csv_table->schema()->ToString() << std::endl;

    // 输出显示Table的数据
    // for (int i = 0; i < csv_table->num_columns(); ++i) {
    //   std::shared_ptr<arrow::Array> column = csv_table->column(i);
    //   std::cout << "Column " << i << ": " << column->ToString() << std::endl;
    // }

    // 1. 显示table信息到std::cout的方法
    // std::shared_ptr<arrow::RecordBatch> record_batch;
    // arrow::Result<std::shared_ptr<arrow::RecordBatch>> result = csv_table->CombineChunksToBatch(); // 执行某个操作，返回Result
    // if (result.ok()) {
    //   record_batch = result.ValueOrDie();
    //   // 在这里使用 record_batch
    // } else {
    //   // 处理错误
    //   std::cerr << "Error: " << result.status().ToString() << std::endl;
    // }
    // //arrow::PrettyPrint(*record_batch, 2, &std::cout);
    // arrow::Status status = arrow::PrettyPrint(*record_batch, 2, &std::cout);
    // if (!status.ok()) {
    //   // 处理错误，例如打印错误信息
    //   std::cerr << "Error: " << status.ToString() << std::endl;
    // }
    // 2. 显示table信息到std::cout的方法
    // std::cout << csv_table->ToString() << std::endl;
    // 3. 显示table信息到std::cout的方法
    // arrow::Status status = arrow::PrettyPrint(*csv_table, 2, &std::cout);
    // if (!status.ok()) {
    //   // 处理错误，例如打印错误信息
    //   std::cerr << "Error: " << status.ToString() << std::endl;
    // }
    // 开始计算夏普率
    // std::cout << "一年的交易日有" << AnnualizationFactors::DAILY << "天" << std::endl;
    // std::cout << DAILY << std::endl;

    // 计算收益率
    arrow::Datum fund_returns;
    arrow::Datum fund_diff;
    std::shared_ptr<arrow::ChunkedArray> cum_nav = csv_table->GetColumnByName("复权净值");
    std::shared_ptr<arrow::ChunkedArray> now_cum_nav = cum_nav->Slice(1,cum_nav->length()-1);
    std::shared_ptr<arrow::ChunkedArray> pre_cum_nav = cum_nav->Slice(0,cum_nav->length()-1);
    ARROW_ASSIGN_OR_RAISE(fund_diff, arrow::compute::CallFunction(
                                          "subtract", {now_cum_nav,pre_cum_nav}));
    ARROW_ASSIGN_OR_RAISE(fund_returns, arrow::compute::CallFunction(
                                          "divide", {fund_diff,pre_cum_nav}));
    // // 获取结果数组
    // std::cout << "Datum kind: " << fund_returns.ToString()
    //           << " content type: " << fund_returns.type()->ToString() << std::endl;

    // // std::cout << fund_returns.scalar_as<arrow::DoubleScalar>().value << std::endl;
    // std::cout << fund_returns.chunked_array()->ToString() << std::endl;
    // 计算夏普率
    arrow::Datum avg_return;
    arrow::Datum avg_std;
    arrow::Datum daily_sharpe_ratio;
    arrow::Datum sharpe_ratio;
    arrow::Datum sqrt_year;
    // 创建 Arrow Double 标量
    double days_of_year_double = 252.0;
    std::shared_ptr<arrow::Scalar> days_of_year = arrow::MakeScalar(days_of_year_double);
    ARROW_ASSIGN_OR_RAISE(sqrt_year, arrow::compute::CallFunction(
                                          "sqrt", {days_of_year}));
    ARROW_ASSIGN_OR_RAISE(avg_return, arrow::compute::CallFunction(
                                          "mean", {fund_returns}));
    arrow::compute::VarianceOptions variance_options;
    variance_options.ddof = 1;
    ARROW_ASSIGN_OR_RAISE(avg_std, arrow::compute::CallFunction(
                                          "stddev", {fund_returns},&variance_options));
    ARROW_ASSIGN_OR_RAISE(daily_sharpe_ratio, arrow::compute::CallFunction(
                                          "divide", {avg_return,avg_std}));
    ARROW_ASSIGN_OR_RAISE(sharpe_ratio, arrow::compute::CallFunction(
                                          "multiply", {daily_sharpe_ratio,sqrt_year}));
    
    std::cout << "计算得到的夏普率为 : " << sharpe_ratio.scalar_as<arrow::DoubleScalar>().value << std::endl;

    auto end_time = std::chrono::high_resolution_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);

    std::cout << "c++读取数据,然后计算夏普率一共耗费时间为: " << duration.count()/1000.0 << " ms" << std::endl;

    return arrow::Status::OK();
  }

// (文档部分: 主函数)
int main() {
  arrow::Status st = RunMain();
  if (!st.ok()) {
    std::cerr << st << std::endl;
    return 1;
  }
  
  return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

CMakeLists.txt

cmake_minimum_required(VERSION 3.16)

project(MyExample)

find_package(Arrow REQUIRED)
find_package(Parquet REQUIRED)
find_package(ArrowDataset REQUIRED)

add_executable(my_example my_example.cc)
target_link_libraries(my_example PRIVATE Arrow::arrow_shared Parquet::parquet_shared ArrowDataset::arrow_dataset_shared)
1
2
3
4
5
6
7
8
9
10

在同一个文件夹下，运行

cmake -B build
cmake --build build
./build/my_example
1
2
3

python 运行代码如下：

import pandas as pd 
import empyrical as ep
import time 
a = time.perf_counter()
data = pd.read_csv("/home/yun/Documents/fund_nav.csv")
returns = data['复权净值'].pct_change().dropna()
sharpe_ratio = ep.sharpe_ratio(returns)
print("计算得到的sharpe_ratio : ", sharpe_ratio)
b = time.perf_counter()
print(f"python读取数据,然后计算夏普率一共耗费时间为: {(b-a)*1000.0} ms")
1
2
3
4
5
6
7
8
9
10