Rust :量化策略回测与tokio异步实践

songroom

426人浏览 · 2025-11-19 15:55:26

songroom · 2025-11-19 15:55:26 发布

在上文中：

Rust : 量化策略回测与多线程技术原型
 Rust: 量化策略回测与简易线程池构建、子线程执行观测
都是标准库中的多线程方案。

当然，需要承认的是，量化策略在很多情况下都是CPU密集型的，其实并不是tokio这种适合IO密集型的最佳场景（也就是没有性能上的提升，甚至某些情况下可能会降低）。

但在一定的程度上，tokio框架和多线程池程池的方案相比，带来了很大程度的简化。

此外，从技术上看，应如何建立一套适用异步的量化策略框架，也值得今天来探讨一下。

一、主要技术要点

除了熟悉Future、async和await基本知识以及tokio库的基本用法外，还需要关注：

1、单个策略的异步抽象化

type Job = Pin<Box<dyn Future<Output=OutPut>  + Send +'static>>;

初步选用了这个方案。这个可以由下面的具体策略抽象化而来：

    let p1 = Parameter::P1(2.0,3.0);
    let s1: Job  = Box::pin(async move { strategy_follow_trend(p1).await});

经过Box::pin后，异步的函数就可以在线程中发送了。

2、策略输入和输出

和多线程，或者同步的方案一致。具体参考上文。

3、设置tokio中多线程参数

我们可以在tokio属性上进行以下设置：

#[tokio::main(flavor = "multi_thread", worker_threads = 3)]

在tokio中，其实可以选单线程和多线程方案，也可以设置具体的线程数量。

而单线程设置模式如下：

#[tokio::main(flavor = "current_thread")]

除了运行时模式，#[tokio::main]还支持其他重要参数有：

worker_threads: 设置工作线程数量（仅适用于multi_thread模式）
start_paused: 启动时暂停计时器（仅适用于current_thread模式）
unhandled_panic: 配置未处理panic的行为（ignore或shutdown_runtime）

通过以上配置，可以满足基本的策略在线程上的需要。

二、代码


use std::thread;
use std::time::{Instant,Duration};
use std::pin::Pin;
#[derive(Debug,Clone)]
enum Parameter{
    P0(()),
    P1(f32,f32),
    P2(f32,f32,f32),
    P3(f32,f32,f32,f32),
}
#[derive(Debug,Clone,PartialEq)]
enum OutPut{
    TradeFlow0(f32),
    TradeFlow1(Vec<f32>,Vec<f32>),
    TradeFlow2(Vec<f32>),
}

async fn strategy_follow_trend(p:Parameter) ->OutPut{
    let thread_id = thread::current().id();
    println!("current id :{:?} run 【follow_trend】 strategy {:?}",thread_id,p);
    //tokio::time::sleep(Duration::from_secs(1)).await;
    thread::sleep(Duration::from_secs(1));
    OutPut::TradeFlow1(vec![1.0,2.0,3.0],vec![4.0,5.0,6.0])
    
}
async fn strategy_bolling(p:Parameter) -> OutPut{
    let thread_id = thread::current().id();
    println!("current id :{:?} run 【bolling】 strategy {:?}",thread_id,p);
    //tokio::time::sleep(Duration::from_secs(1)).await;
    thread::sleep(Duration::from_secs(1));
    OutPut::TradeFlow2(vec![1.0,2.0,3.0])
    
}
async fn strategy_high_freq(p:Parameter)->OutPut{
    let thread_id = thread::current().id();
    println!("current id :{:?} run 【high_freq】 strategy {:?}",thread_id,p);
    //tokio::time::sleep(Duration::from_secs(1)).await;
    thread::sleep(Duration::from_secs(1));
    
    OutPut::TradeFlow0(0.0)
    
}

fn get_strategies() -> Vec<Job>{
    let p1 = Parameter::P1(2.0,3.0);
    let s1: Job  =Box::pin(async move { strategy_follow_trend(p1).await});
    let p2 = Parameter::P2(2.0,3.0,4.0);
    let s2: Job  = Box::pin(async move { strategy_bolling(p2).await});
    let p3 = Parameter::P0(());
    let s3: Job  = Box::pin(async move { strategy_high_freq(p3).await});
    let p4 = Parameter::P0(());
    let s4: Job  = Box::pin(async move { strategy_high_freq(p4).await});
    let p5 = Parameter::P1(2.0,3.0);
    let s5: Job  = Box::pin(async move { strategy_follow_trend(p5).await});
    let p6 = Parameter::P2(2.0,3.0,4.0);
    let s6: Job  = Box::pin(async move { strategy_bolling(p6).await});
    let p7 = Parameter::P0(());
    let s7: Job  = Box::pin(async move { strategy_high_freq(p7).await});    
    let p8 = Parameter::P0(());
    let s8: Job  = Box::pin(async move { strategy_high_freq(p8).await});
    let strategies:Vec<Job> = vec![s1,s2,s3,s4,s5,s6,s7,s8];
    strategies
}

// 定义闭包中没有参数输入的函数类型，做为发送对象
// 需要加Send约束的原因是便于在线程中发送，支持Spawn函数
type Job = Pin<Box<dyn Future<Output=OutPut>  + Send +'static>>;

#[tokio::main(flavor = "multi_thread", worker_threads = 3)]
async fn main(){
    let start = Instant::now();
    let jobs:Vec<Job> = get_strategies();
    let strategy_num = jobs.len();
    let mut handles = Vec::new();
    
    for job in jobs{
        let handle = tokio::spawn(async move{
            let _ = job.await;
        });
        handles.push(handle);
    }
    for handle in handles{
        handle.await.unwrap();
    }
    
    let duration = start.elapsed();
    println!("strategy num : {:?}",strategy_num);
    println!("all strategy 运行完毕! duration : {:?}",duration);
}

三、输出

我们在上面设置了多线程，线程数量是3，运行结果如下：

current id :ThreadId(2) run 【follow_trend】 strategy P1(2.0, 3.0)
current id :ThreadId(3) run 【high_freq】 strategy P0(())
current id :ThreadId(4) run 【bolling】 strategy P2(2.0, 3.0, 4.0)
current id :ThreadId(4) run 【high_freq】 strategy P0(())
current id :ThreadId(3) run 【follow_trend】 strategy P1(2.0, 3.0)
current id :ThreadId(2) run 【bolling】 strategy P2(2.0, 3.0, 4.0)
current id :ThreadId(4) run 【high_freq】 strategy P0(())
current id :ThreadId(2) run 【high_freq】 strategy P0(())
strategy num : 8
all strategy 运行完毕! duration : 3.0024845s