Rust : 量化策略回测与多线程技术原型

songroom

510人浏览 · 2025-11-11 09:21:45

songroom · 2025-11-11 09:21:45 发布

在Python中，经常可能会用单线程来执行多策略的回测，这样效率会低一些。

在Rust中，用多线程来同时跑多策略，榨干机器的性能，提高回测效率上是很方便的。
但不同的策略的输入和输出，以及函数的签名有所不同，如果放在Rust中多线程中运行，需要提前进行一些处理。

下面主要探讨上面的一些技术要点：

一、主要技术要点

1、输入参数封装

在策略的输入上，我们不妨把形形色色的策略参数放到一个enum结构中，进行统一封装。此外，如果有新策略参数，只需要不断维护就可以了。

#[derive(Debug,Clone)]
enum Parameter{
    P0(()),//无参数
    P1(f32,f32),//两个参数
    P2(f32,f32,f32),//三个参数
    P3(f32,f32,f32,f32),//四个参数，......可以自定义设置
}

2、策略输出封装

在策略输出上，也可以进行自定义。比如输出是交易信号。下面只是一个举例。

#[derive(Debug,Clone,PartialEq)]
enum OutPut{
    TradeFlow0(f32),
    TradeFlow1(Vec<f32>,Vec<f32>),
    TradeFlow2(Vec<f32>),
}

当然，更真实的场景可能是：

use chrono::NaiveDateTime;
use std::collections::HashMap;
#[derive(Debug,Clone,PartialEq)]
enum Parameter{
    P0(StratgegyA),
    p1(StratgegyB),
    p3(StratgegyC),
    p4(StratgegyD),
}
#[derive(Debug,Clone,PartialEq)]
struct StratgegyA{
    p1:f32,
    p2:f32,
}
#[derive(Debug,Clone,PartialEq)]
struct StratgegyB{
    p1:f32,
    p2:f32,
    p3:f32,
}
#[derive(Debug,Clone,PartialEq)]
struct StratgegyC{
    p1:f32,
    p2:f32,
    p3:f32,
    p4:f32,
}
#[derive(Debug,Clone,PartialEq)]
struct StratgegyD{
    p1:f32,
    p2:f32,
    p3:f32,
    p4:f32,
}

#[derive(Debug,Clone,PartialEq)]
enum OutPut{
    TradeFlow(HashMap<String,TradeFlow>),//交易代码+交易成交流水
    TradeFlow1(HashMap<String,Vec<TradeFlow>>),//策略名+交易流水
    TradeFlow2(HashMap<String,TradeFlow2>),
    TradeFlow3(Vec<TradeFlow>),//交易流水
}
// 记录单次交易流的状态
#[derive(Debug,Clone,PartialEq)]
struct TradeFlow2{
    strategy_name: String,
    trade_id : i32,
    symbol:String,
    trade_status:i8,//1:open, 2:close
    direction:i8,//1:long, -1:short
    trade_price:f32,
    trade_time:NaiveDateTime,
    trade_volume:f32,
    trade_amount:f64
    trade_fee:f32,
    available_fund:f32
}
//记当交易对开平的完整状态
#[derive(Debug,Clone,PartialEq)]
struct TradeFlow{
    strategy_name: String,
    flow_id : i32,
    symbol:String,
    open_price:f32,
    open_time:NaiveDateTime,
    close_price:f32,
    close_time:NaiveDateTime,
    open_direction:i8,//1:long, -1:short
    trade_volume:f32,
    trade_duration:Duration,
    trade_fee:f32,
    available_fund:f32
}

这里只是根据模拟的需要，进行了相应的简化。

3、策略函数的封装

根据上面的输入和输出，对于不同的策略签名：

strategy_fn(Parameter)->Output

已经抽象成上面的样子。

我们再放到闭包中即：

move ||strategy_fn(Parameter)->Output

就会抽象成统一的格式，即

FnOnce()->OutPut

再进行Box打包，就完成了策略的统一化封装。

具体如下：

type Job = Box<dyn FnOnce()->OutPut + Send +'static>;

上面有两个问题：

第1、为什么需要加Send？这样是为了满足多线程Spawn函数Send约束的需要。

第2、为什么是FnOnce()？不是明明这些策略都有参数嘛，因为我们在“||”中并没有参数，故是FnOnce（）。

4、多线程

直接可以调用标准库成熟的多线程thread中Spawn函数，然后再join一下，即可。在这里选择1：1模式，即1个策略对应1个线程来进行模拟。

二、具体代码

根据上面的思路，全部代码如下：

use std::sync::{mpsc,Arc,Mutex};
use std::thread;
use std::time::{Instant,Duration};
#[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>),
}
fn process<Strategy>(p:Parameter,s:Strategy) -> OutPut  where Strategy: FnOnce(Parameter)->OutPut {
    s(p)
}
fn test() -> OutPut{
    process(Parameter::P0(()),|p|strategy_follow_trend(p))
}
fn strategy_follow_trend(p:Parameter) ->OutPut{
    println!("run 【follow_trend】 strategy {:?}",p);
    thread::sleep(Duration::from_secs(1));
    OutPut::TradeFlow1(vec![1.0,2.0,3.0],vec![4.0,5.0,6.0])
    
}
fn strategy_bolling(p:Parameter) -> OutPut{
    println!("run 【bolling】 strategy {:?}",p);
    thread::sleep(Duration::from_secs(1));
    OutPut::TradeFlow2(vec![1.0,2.0,3.0])
    
}
fn strategy_high_freq(p:Parameter)->OutPut{
    println!("run 【high_freq】 strategy {:?}",p);
    thread::sleep(Duration::from_secs(1));
    OutPut::TradeFlow0(0.0)
    
}
#[test]
pub fn test_strategy(){
    // 测试follow_trend策略
    let p1 = Parameter::P1(2.0,3.0);
    let out = strategy_follow_trend(p1);
    assert_eq!(out,OutPut::TradeFlow1(vec![1.0,2.0,3.0],vec![4.0,5.0,6.0]));
    //格式验证
    assert_eq!(test(),OutPut::TradeFlow1(vec![1.0,2.0,3.0],vec![4.0,5.0,6.0]));
}
fn get_strategies() -> Vec<Job>{
    let p1 = Parameter::P1(2.0,3.0);
    let s1: Job  = Box::new(move || strategy_follow_trend(p1));
    let p2 = Parameter::P2(2.0,3.0,4.0);
    let s2: Job  = Box::new(move || strategy_bolling(p2));
    let p3 = Parameter::P0(());
    let s3: Job  = Box::new(move || strategy_high_freq(p3));
    let p4 = Parameter::P0(());
    let s4: Job  = Box::new(move || strategy_high_freq(p4));
    let p5 = Parameter::P1(2.0,3.0);
    let s5: Job  = Box::new(move || strategy_follow_trend(p5));
    let p6 = Parameter::P2(2.0,3.0,4.0);
    let s6: Job  = Box::new(move || strategy_bolling(p6));
    let p7 = Parameter::P0(());
    let s7: Job  = Box::new(move || strategy_high_freq(p7));
    let p8 = Parameter::P0(());
    let s8: Job  = Box::new(move || strategy_high_freq(p8));
    let strategies:Vec<Job> = vec![s1,s2,s3,s4,s5,s6,s7,s8];
    strategies
}
// 定义闭包中没有参数输入的函数类型，做为发送对象
// 需要加Send约束的原因是便于在线程中发送，支持Spawn函数
type Job = Box<dyn FnOnce()->OutPut + Send +'static>;
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 = thread::spawn(move ||{
             job();
        });
        handles.push(handle);   
    }
    
    for handle in handles{
        handle.join().unwrap();
    }
    let duration = start.elapsed();
    println!("strategy num : {:?}",strategy_num);
    println!("all strategy 运行完毕! duration : {:?}",duration);
}

三、技术验证

不管是cargo test，还是cargo run，运行结果和预期一致。输出如下：

run 【bolling】 strategy P2(2.0, 3.0, 4.0)
run 【follow_trend】 strategy P1(2.0, 3.0)
run 【high_freq】 strategy P0(())
run 【high_freq】 strategy P0(())
run 【follow_trend】 strategy P1(2.0, 3.0)
run 【high_freq】 strategy P0(())
run 【bolling】 strategy P2(2.0, 3.0, 4.0)
run 【high_freq】 strategy P0(())
strategy num : 8
all strategy 运行完毕! duration : 1.0031163s