# mdtx-online

**Repository Path**: gitfun/mdtx-online

## Basic Information

- **Project Name**: mdtx-online
- **Description**: Online algorithms used in mdtx
- **Primary Language**: C++
- **License**: BSD-3-Clause
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 3
- **Created**: 2022-07-28
- **Last Updated**: 2022-07-28

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# mdtx-online

#### 介绍

mdtx-online是研究/回测/交易系统mdtx的一部分，旨在提供一个简单且逻辑统一的在线计算框架。提供的算法多数提供O(1)更新，涉及排序的算法一般提供O(log N)，
基本可以满足开发高性能事实策略。

#### 前言

在日常交流中，我发现多数同学们对交易数据的处理仍处于每日一次“拉取当天数据——计算相关指标——执行策略回测”的离线模式。这种思路足够成熟鲁棒，一般情况下可以
满足多数日线级别量化交易需求。如比较流行的ta-lib库，就只能适合上述离线计算。然而，这种计算框架往往无法满足想要进一步迈入日内交易领域的同学。mdtx-online
正是为了解决在线计算工具的缺失而存在的。

#### 在线计算的逻辑

指标计算逻辑一般分为两种：累积计算和滚动计算：累积计算函数应用于当前所有数据点，如CMA，计算当前所有数据点算术平均；滚动计算函数应用于滚动窗口内的数据点，
如SMA(10)，计算最近10个数据点的算数平均。mdtx-online的更新器（updater/*）对累积计算和滚动计算不做区分。对所有到达系统的数据点，更新器可以进行插入
（insert），移除（remove）和查询（get/pop）操作。与通用流计算算法（如RA，Two-stacks，DABA等）不同，更新器只保存当前状态而并不维护全部窗口，因此用
只能查询更新器最新的状态，而无法通过combine操作查询历史状态。采用这个设计的原因有：

    1. 日内交易算法一般只关注最新状态。

    2. O(1)的时间与空间复杂度。更低的延迟。

    3. 实时金融数据一般情况下严格FIFO，实时算法设计上一般无需考虑迟到数据与水印问题。


总而言之，在线计算的逻辑被简单抽象为：获得新数据 —— 把新数据插入到更新器中（insert） —— 滚动计算？ —— 把旧数据从更新器中移除（remove）—— 查询结果（get）

所有的更新器都继承自模版mdtx::online::update::update_base，该模版提供一些更新器的信息如：

    * scala_in/scale_out 基础数据类型

    * tuple_in/tuple_out 输入输出元组类型

    * num_in/num_out 输入输出数据点数量

同时定义了通用方法：

    * virtual void insert(tuple_in const &) 插入数据点

    * virtual void remove(tuple_in const &) 移除数据点

    * virtual void reset() 重置更新器

    * virtual tuple_out get() 查询更新器

    * tuple_out pop() 查询并重置更新器

如 ```updater::argminmax<double>``` 输入基础数据类型是double，输出基础数据类型是ptrdiff_t，更新器接受的输入元组是```std::tuple<double>```，输出元组类
型是```std::tuple<ptrdiff_t, ptrdiff_t>```（因为num_in=1， num_out=2）。

例1：使用更新器在线计算过去3秒内报价的方差：

```C++
#include <tuple>
#include <boost/circular_buffer.hpp>
#include "mdtx-online/updater/s2.h"

using namespace mdtx::online;

class roll_stddev
{
private:
    using time_datum = std::tuple<int, double>;

    int window; // 时间窗口大小

    boost::circular_buffer<time_datum> buffer; // 数据缓存
    updater::var<double> s2{1}; // 方差更新器，构造函数参数ddof=1，计算样本方差

public:
    explicit roll_stddev(int window = 3, int init_buf = 32)
        : window(window),
          buffer(init_buf)
    {
    }

    double on_data(int timestamp, double price) // 新数据到达
    {
        var.insert(price); // 插入新数据点

        int pop = 0; // 记录过期数据点数量
        int barrier = timestamp - window; // 小于barrier的均为过期数据点，需要移除
        for (auto const &[time, val] : buffer)
        {
            if (time >= barrier) // 因为行情数据确保FIFO，当前数据点未过期时，提前结束循环
                break;
            var.remove(val); // 从更新器中移除过期数据点
            ++pop; // 记录过期数据点数量
        }
        buffer.erase_begin(pop); // 从缓存中移除全部过期数据点

        if (buffer.full()) // 防止缓存溢出
        {
            buffer.set_capacity(buffer.size() * 2);
        }
        buffer.push_back(time_datum(time, price)); // 记录最新数据点到缓存中

        auto [var] = s2.get(); // 查询最新方差，因为输出是std::tuple<double>，此处直接解包
        return sqrt(var); // 标准差 = sqrt(方差)
    }
};
```

例3：我们来设计一个玩具指标，计算最近X手成交量内，最大值成交价偏离标准差的程度。我们希望有个量化指标可以体现窗口内最高成交价对比平均成交价偏离程度的大小，并以此来衡量平均回归强度。为了达成这个计算，我们需要滚动计算窗口内的：1. 标准差， 2. 平均值， 3. 最大值。

```C++
#include <tuple>
#include <boost/circular_buffer.hpp>
#include "mdtx-online/updater/s2.h"
#include "mdtx-online/updater/minmax.h"

using namespace mdtx::online;

class roll_maxscore
{
private:
    double volume_window;
    updater::sum<double> vol{};
    updater::var<double> s2{1};
    updater::max<double> max{};
    boost::circular_buffer<double> buf;

public:
    explicit roll_maxscore(double volume_window_size, int init_buf = 32)
        : volume_window(volume_window_size),
          buf(init_buf + init_buf % 2) // 确保缓存大小是偶数
    {
    }

    double on_data(double price, double volume)
    {
        // 插入新数据点
        vol.insert(volume);
        var.insert(price);
        max.insert(price);

        // 缓存模式（从旧到新）
        // |volume 0|price 0|volume 1|price 1|......|volume n|price n|
        while (!buf.empty() && vol.s > volume_window) // vol是简单累加器，vol.s是当前总和
        {
            // 移除过期数据点
            vol.remove(buf[0]);
            var.remove(buf[1]);
            max.remove(buf[1]);
            buf.erase_front(2);
        }

        if (buffer.full()) // 防止缓存溢出
        {
            buffer.set_capacity(buffer.size() * 2);
        }
        buf.push_back(volume);
        buf.push_back(price)

        auto [high] = max.get();
        auto [s2] = var.get();
        double delta = high - var.m; // 方差更新器同时会维护平均值，m，无需重复计算
        return delta / sqrt(s2); // 计算偏离标准分
    }
};
```

#### 使用indicator类

上节阐述了mdtx-online的在线计算逻辑，使用更新器类可以轻松构建高实时性的在线算法。然而，更新器只提供了基础的计算模块，使用仍不够方便。所以mdtx-online还提供
了indicator类，indicator类基于更新器，实现了许多常用指标的在线计算版本。从而用户无需重复发明轮子。

indicator类均为仿函数（提供operator()）方法。所有indicator类都提供单点计算（update, 适合日内的实时计算），和批量计算（update_n, 适合回测环境）。请参考下例和C++文档使用。

```C++
#include <vector>
#include <algorithm>
#include "mdtx-online/indicator/ma.h"
#include "mdtx-online/indicator/ma_deriv.h"

namespace indi = mdtx::online::indicator;

// 计算成交量加权MACD作为操作指标

class dummy_strategy
{
    // 构造一个计算量加权移动平均背离的macd
    using my_macd = indi::macd_wrapper<indi::sma_vw<double>>;

    my_macd macd;

public:
    dummy_strategy(int fast, int slow, int sig)
        : macd(fast, slow, sig)
    {
    }

    // 单点计算模式，适合实时计算指标信号
    int on_bar(double price, double volume)
    {
        // 计算成交量加成macd，ma/sig/hist为macd的输出数据
        auto [ma, sig, hist] = macd(price, volume);
        if (sig > 0 && hist > 0)
        {
            return 1; // 买入
        }
        else if (sig < 0 && hisg < 0)
        {
            return -1; // 卖出
        }
        else
        {
            return 0; // 不操作
        }
    }

    // 批量计算模式，适合一次计算全部指标进行回测分析
    std::vector<int> run(const std::vector<double> &price, const std::vector<double> &volume)
    {
        size_t n = price.size();
        std::vector<double> ma(n), sig(n), hist(n); // 预分配空间，也可使用back_inserter
        std::vector<double> result(n); // 预分配空间
        // 批量计算macd
        macd.update_n(n, price.begin(), volume.begin(), ma.begin(), sig.begin(), hist.begin());
        // 批量计算信号
        std::transform(sig.begin(), sig.end(), hist.begin(), result.begin(),
        [](auto signal, auto histo)
        {
            if (signal > 0 && histo > 0)
            {
                return 1;
            }
            else if (signal < 0 && histo < 0)
            {
                return -1;
            }
            else
            {
                return 0;
            }
        });
        return result;
    }
}

int main()
{
    ...
    ...
    ...

    strat = dummy_strategy(20, 60, 15);
    // 主事件循环
    while (event_loop)
    {
        auto [price, volume] = get_lastest_bar(); // 获取最新价格
        auto signal = strat.on_bar(price, volume); // 计算策略信号
        if (signal > 0)
        {
            buy(); // 买入信号
        }
        else if (signal < 0)
        {
            sell(); // 卖出信号
        }
    }

    ...
    ...
    ...
}

```

##### 关于性能

mdtx-online中多数算法只需O(1)更新，且circular_buffer较为高效，因此非专业机构用户在使用mdtx-online构建日内策略时，瓶颈往往是在数据IO，用户无需担心策略运行
效率。若用户希望实现复杂高性能策略，甚至高频策略，我不建议使用mdtx-online中的默认indicator实现。原因是每个indicator类都维护自己的私有环形缓冲，因此有额外的
内存和寻址开销。为进一步提高策略性能，用户应使用更新器共用缓冲构建策略。container中提供的排序容器可实现每秒百万次查找插入操作，用户可配合实现较为复杂的高性能实时
策略。

#### 采样器

市场数据是离散且往往不等宽等频，因子直接使用原始数据进行计算数学上可能不尽合理。mdtx-online提供实时聚合器来帮助有需求的用户对实时数据进行聚合。sampler使用与indicator类似，也提供单点计算与批量计算接口。请参考C++文档使用。

```C++

#include "mdtx-online/sampler/tumbling.h"
#include "mdtx-online/sampler/downsample.h"

using namespace mdtx::online;

int main()
{
    ...
    ...
    ...

    sampler::volclk<int, double> vclk(volume = 1000.0); // 量钟, 1000股tumbling采样
    sampler::tumbling<int, double> kbar(window = 30); // 30秒k线，时间使用int类型
    sampler::downsample_tumbling<int, double> kbar5min(window = 300) // 5分钟k线，重采样

    while (event_loop)
    {
        auto [time, price, volume, turnover] = get_latest_data();
        // vbar是std::optional<bar_type>
        // 其中bar_type是包含时间，开，高，低，收，量，换手，均价的元组
        auto vbar = vclk(time, price, volume, turnover);
        if (vbar)
        {
            auto [tnow, o, h, l, c, v, tnvr, vwap] = *vbar;
            // 处理量钟信号
            on_vbar1000(tnow, o, h, l, c, v, tnvr, vwap);
            ...
            ...
        }
        auto bar = kbar(time, price, volume, turnover);
        if (bar)
        {
            auto [tnow, o, h, l, c, v, tnvr, vwap] = *bar;
            // 处理30秒信号
            on_bar30s(tnow, o, h, l, c, v, tnvr, vwap);
            ...
            ...
        }
        if (bar)
        {
            // 重采样低频数据
            auto bar5 = kbar5min(*bar);
            if (bar5)
            {
                auto [tnow, o, h, l, c, v, tnvr, vwap] = *bar;
                // 处理5分钟低频信号
                on_bar5min(tnow, o, h, l, c, v, tnvr, vwap);
                ...
                ...
                ...
            }
        }
    }

}

```

#### 语言绑定

mdtxr是mdtx-online的R语言绑定，manano是mdtx-online的Python语言绑定

#### 安装教程

函数库只包含头文件，直接 #include 所需算法头文件使用。

#### 参与贡献

1.  Fork 本仓库
2.  新建分支
3.  提交代码
4.  新建 Pull Request