indicator_set/src/msv.rs

use std::cmp::max;
use std::collections::BTreeMap;
use std::str::FromStr;
use actix_web::{HttpResponse};
use chrono::Utc;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use serde_json::{json, Value};
use crate::db_connector::get_trades_json;
use crate::server::{Response, 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 query_value["exchange"].as_str() {
        None => {
            "gate_usdt_swap"
        }
        Some(exchange) => {
            exchange
        }
    };
    let symbol = match query_value["symbol"].as_str() {
        None => {
            "BTC_USDT".to_string()
        }
        Some(symbol) => {
            symbol_fix(symbol)
        }
    };
    let mills_back = match query_value["mills_back"].as_str() {
        None => {
            dec!(37)
        }
        Some(mills_back_str) => {
            Decimal::from_str(mills_back_str).unwrap()
        }
    };
    let minute_time_range = match query_value["minute_time_range"].as_str() {
        None => {
            240
        }
        Some(minute_time_range_str) => {
            match i64::from_str(minute_time_range_str) {
                Ok(minute_time_range) => {
                    minute_time_range
                }
                Err(_) => {
                    // 返回数据
                    let response = Response {
                        query: query_value.clone(),
                        msg: Some("时间不要字符串，要数字，这个版本不支持字符串".to_string()),
                        code: 500,
                        data: Value::Null,
                    };

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

    // 链接数据服务器查询数据
    let end_time = Utc::now().timestamp_millis();
    let start_time = end_time - minute_time_range * 60 * 1000;
    let db_response = get_trades_json(
        exchange,
        symbol.as_str(),
        start_time,
        end_time,
    ).await;

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

        // 返回数据
        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(&db_response).unwrap();
        HttpResponse::BadRequest().content_type("application/json").body(json_string)
    }
}

// 将trades转换为具体指标
pub fn generate_msv_by_trades(mut trades: Vec<Trade>, mills_back: Decimal) -> Value {
    let mut amplitude_map: BTreeMap<Decimal, Decimal> = BTreeMap::new();

    // 每一个元素都遍历一遍
    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 max_price = dec!(-1);
        let mut min_price = dec!(1e28);

        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;
            }

            // 判断最大值、最小值
            if flag_trade.price > max_price {
                max_price = flag_trade.price;
            }
            if flag_trade.price < min_price {
                min_price = flag_trade.price;
            }

            range_index -= 1;
        }

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

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

            up_rate.rescale(2);
            dn_rate.rescale(2);

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

            // 如果已经生成了一个波动，则也要和已生成的波动进行比较
            let insert_value = if amplitude_map.contains_key(&trade.time) {
                let origin_rate = amplitude_map.get(&trade.time).unwrap();

                if up_rate > dn_rate.abs() {
                    max(*origin_rate, up_rate)
                } else {
                    max(*origin_rate, dn_rate)
                }
            } else {
                if up_rate > dn_rate.abs() {
                    up_rate
                } else {
                    dn_rate
                }
            };

            amplitude_map.insert(trade.time, insert_value);
        }
    }

    let x: Vec<Decimal> = amplitude_map.keys().cloned().collect();
    let y: Vec<Decimal> = amplitude_map.values().cloned().collect();
    let total_size = trades.len();
    let result_size = x.len();
    json!({
        "x": x,
        "y": y,
        "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
}