indicator_set/src/msv.rs

use std::cmp::{max, min};
use std::str::FromStr;
use actix_web::{HttpResponse};
use chrono::Utc;
use rust_decimal::Decimal;
use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
use rust_decimal_macros::dec;
use serde_json::{json, Value};
use crate::db_connector::{get_simple_depths_json, get_trades_json};
use crate::params_utils::{get_str, parse_str_to_decimal};
use crate::server::{Response, SimpleDepth, Trade};

pub fn symbol_fix(symbol: &str) -> String {
    let mut fixed = symbol.to_uppercase();

    if !fixed.contains("_USDT") {
        fixed = format!("{}_USDT", fixed);
    }

    fixed
}

// 将trades_json转换为指标
pub async fn generate_msv(query_value: Value) -> HttpResponse {
    // 参数处理
    let exchange = match get_str(query_value.clone(), "exchange") {
        Ok(str) => {
            str
        }
        Err(response) => {
            return response
        }
    };
    let symbol = match get_str(query_value.clone(), "symbol") {
        Ok(symbol) => {
            symbol_fix(symbol.as_str())
        }
        Err(response) => {
            return response
        }
    };
    let mills_back = match parse_str_to_decimal(query_value.clone(), "mills_back") {
        Ok(t) => {
            t
        }
        Err(response) => {
            return response
        }
    };
    let minute_time_range = match parse_str_to_decimal(query_value.clone(), "minute_time_range") {
        Ok(t) => {
            t.to_i64().unwrap()
        }
        Err(response) => {
            return response
        }
    };

    // 链接数据服务器查询数据
    let end_time = Utc::now().timestamp_millis();
    let start_time = end_time - minute_time_range * 60 * 1000;
    let trades_response = get_trades_json(
        exchange.as_str(),
        symbol.as_str(),
        start_time,
        end_time,
    ).await;
    let simple_start_time = if start_time < end_time - 240 * 60 * 1000 {
        end_time - 240 * 60 * 1000
    } else {
        start_time
    };
    let simple_depths_response = get_simple_depths_json(
        exchange.as_str(),
        symbol.as_str(),
        simple_start_time,
        end_time,
    ).await;

    // 对数据库返回的数据进行容错处理
    if trades_response.code == 200 && simple_depths_response.code == 200 {
        // 数据本地化处理
        let trades = parse_json_to_trades(trades_response.data);
        let simple_depths = parse_json_to_simple_depths(simple_depths_response.data);
        // 指标生成
        let indicator = generate_msv_by_trades(trades, mills_back, simple_depths, start_time, end_time);

        // 返回数据
        let response = Response {
            query: query_value.clone(),
            msg: Some("指标生成完毕".to_string()),
            code: 200,
            data: indicator,
        };

        let json_string = serde_json::to_string(&response).unwrap();
        HttpResponse::Ok().content_type("application/json").body(json_string)
    } else {
        let json_string = serde_json::to_string(&trades_response).unwrap();
        HttpResponse::Ok().content_type("application/json").body(json_string)
    }
}

// 将trades转换为具体指标
pub fn generate_msv_by_trades(mut trades: Vec<Trade>, mills_back: Decimal, simple_depths: Vec<SimpleDepth>, start_time: i64, end_time: i64) -> Value {
    let mut msv_data: Vec<Vec<Decimal>> = vec![];

    const GAMMA: Decimal = dec!(0.5);

    // 每一个元素都遍历一遍
    trades.sort_by(|a, b| Decimal::from_str(a.id.as_str()).unwrap().cmp(&Decimal::from_str(b.id.as_str()).unwrap()));
    for (index, trade) in trades.iter().enumerate() {
        // 该元素向前遍历range毫秒
        let mut range_index = if index == 0 {
            0
        } else {
            index
        };

        // 计算区间的预定价格
        let mut ref_price = trade.price;
        let mut dissociation = Decimal::ZERO;
        loop {
            // 第0个就不搞
            if range_index == 0 {
                break;
            }

            let flag_trade = trades.get(range_index).unwrap();
            let range_time = trade.time - flag_trade.time;
            // 判断该ticker是否是range ms以外
            if range_time > mills_back {
                break;
            }

            ref_price = ref_price * GAMMA + flag_trade.price * (Decimal::ONE - GAMMA);
            dissociation = dissociation + flag_trade.size.abs();

            range_index -= 1;
        }

        // 逻辑计算层
        // 取离当前点最远的点进行测量
        let last_price = trade.price;

        // 不是初始值，以及不是0波动
        if index != 0 {
            let mut rate = Decimal::ONE_HUNDRED * (last_price - ref_price) / ref_price;

            rate.rescale(2);

            // 去除小数位之后，可以忽略一些太小的波动，减少图表生成压力
            if rate.eq(&Decimal::ZERO) {
                continue
            }

            // 去重，以及保留最大的波动率
            if msv_data.len() > 0 {
                let last = msv_data[msv_data.len() - 1].clone();
                let last_time = last[0];
                let last_rate = last[1];

                // 如果时间相同，则可能会进行remove等操作
                if last_time == trade.time {
                    // 如果最新的波动率大于最后波动率
                    if rate.abs() > last_rate.abs() {
                        msv_data.remove(msv_data.len() - 1);
                        msv_data.push(vec![trade.time, rate, dissociation]);
                    }
                } else {
                    msv_data.push(vec![trade.time, rate, dissociation]);
                }
            } else {
                msv_data.push(vec![trade.time, rate, dissociation]);
            }
        }
    }

    // 按时间序列填充数据
    let mut msv_index = 0;
    let mut final_msv_data: Vec<Vec<Decimal>> = vec![];

    let mut depth_index = 0;
    let mut final_depth_data: Vec<Vec<Decimal>> = vec![];

    let mut final_spread_data: Vec<Vec<Decimal>> = vec![];

    let mut index_timestamp = Decimal::from_i64(start_time).unwrap();
    let last_timestamp = Decimal::from_i64(end_time).unwrap();
    let step_timestamp = dec!(1000);
    loop {
        let mut max_msv_data = Decimal::ZERO;
        let mut max_msv_diss_data = Decimal::ZERO;

        // ====================================== 数据生产 ===============================================
        // 获取时间范围内的波动率数据
        loop {
            // 下标合法性判断
            if msv_index >= msv_data.len() {
                break;
            }

            // msv_data的指定下标数据不在时间范围内(时间范围：指的是[index_timestamp-mills_back, index_timestamp]这个范围)
            if index_timestamp < msv_data[msv_index][0] {
                break;
            }

            // -------------- 大小判断，取值
            let msv_d = msv_data[msv_index][1];
            let msv_diss_data = msv_data[msv_index][2];
            // msv波动数据
            if max_msv_data.abs() < msv_d.abs() {
                max_msv_data = msv_d;
            }
            // 堆叠的交易量数据
            if max_msv_diss_data < msv_diss_data {
                max_msv_diss_data = msv_diss_data;
            }
            // 下标步近
            msv_index = msv_index + 1;
        }

        // 获取时间范围内的深度数据、买一及卖一价数据
        let mut max_size = Decimal::ZERO;
        let mut min_size = Decimal::ONE_THOUSAND * Decimal::ONE_THOUSAND;
        let mut max_spread = Decimal::ZERO;
        loop {
            // 下标合法性判断
            if depth_index >= simple_depths.len() {
                break;
            }
            let depth = &simple_depths[depth_index];
            // 时间范围合法性判断，只统计那一秒以内的深度总交易量
            if index_timestamp < depth.time {
                break;
            }
            // 这一秒的深度最大值、最小值
            max_size = max(max_size, depth.size);
            min_size = min(min_size, depth.size);
            // 这一秒的差价最大值
            let spread = depth.a1 - depth.b1;
            max_spread = max(spread, max_spread);
            // 下标步近
            depth_index += 1;
        }

        // ====================================== 智能填充数据 ===============================================
        // 价差
        final_spread_data.push(vec![index_timestamp, max_spread]);

        // 流动性数据叠加
        let rst_size = if (max_size == Decimal::ZERO || min_size == Decimal::ONE_THOUSAND * Decimal::ONE_THOUSAND) && final_depth_data.len() > 0 {
            final_depth_data.last().unwrap()[1]
        } else {
            if (max_size == Decimal::ZERO || min_size == Decimal::ONE_THOUSAND * Decimal::ONE_THOUSAND) && simple_depths.len() > 0 {
                simple_depths[0].size
            } else {
                if simple_depths.len() > 0 {
                    (max_size + min_size) / Decimal::TWO
                } else {
                    Decimal::ZERO
                }
            }
        };
        final_depth_data.push(vec![index_timestamp, rst_size]);

        // 波动率数据处理
        // 如果这两个值为0，则代表这mills_back毫秒以内是没有数据的，填充0数据，使得x轴是完整的
        if max_msv_data == Decimal::ZERO {
            final_msv_data.push(vec![index_timestamp, Decimal::ZERO, Decimal::ZERO]);

        // 说明在这个时间范围内是有数据存在的
        } else {
            final_msv_data.push(vec![index_timestamp, max_msv_data, max_msv_diss_data]);
        }

        // ====================================== 时间步进处理 ======================================
        // 对时间进行步近
        index_timestamp = index_timestamp + step_timestamp;
        // 时间越界
        if index_timestamp > last_timestamp {
            break
        }
    }

    // 结果统计
    let total_size = trades.len();
    let result_size = final_msv_data.len();
    json!({
        "msv": final_msv_data,
        "liqs": final_depth_data,
        "spreads": final_spread_data,
        "total_size": total_size,
        "result_size": result_size,
    })
}

// 将json转换为trades
pub fn parse_json_to_trades(trades_json: Value) -> Vec<Trade> {
    let mut rst = vec![];

    for trade_json in trades_json.as_array().unwrap() {
        let arr = trade_json.as_array().unwrap();
        rst.push(Trade {
            id: arr[0].as_str().unwrap().to_string(),
            time: Decimal::from_str(arr[1].as_str().unwrap()).unwrap(),
            size: Decimal::from_str(arr[2].as_str().unwrap()).unwrap(),
            price: Decimal::from_str(arr[3].as_str().unwrap()).unwrap(),
        });
    }

    rst
}

// 将json转换为简易深度数据
pub fn parse_json_to_simple_depths(depths_json: Value) -> Vec<SimpleDepth> {
    let mut rst = vec![];

    for depth_json in depths_json.as_array().unwrap() {
        let depth: SimpleDepth = SimpleDepth {
            time: Decimal::from_str(depth_json["time"].as_str().unwrap()).unwrap(),
            size: Decimal::from_str(depth_json["size"].as_str().unwrap()).unwrap(),
            a1: Decimal::from_str(depth_json["a1"].as_str().unwrap()).unwrap(),
            b1: Decimal::from_str(depth_json["b1"].as_str().unwrap()).unwrap(),
        };

        rst.insert(0, depth)
    }

    rst
}