crypto_quant/market_data_main.py

import core.logger as logging
from datetime import datetime
from time import sleep
import pandas as pd
from core.biz.market_data import MarketData
from core.db.db_market_data import DBMarketData
from core.biz.metrics_calculation import MetricsCalculation
from core.utils import (
    datetime_to_timestamp,
    timestamp_to_datetime,
    transform_date_time_to_timestamp,
)
from trade_data_main import TradeDataMain
from config import (
    API_KEY,
    SECRET_KEY,
    PASSPHRASE,
    SANDBOX,
    MONITOR_CONFIG,
    MYSQL_CONFIG,
    BAR_THRESHOLD,
)

logger = logging.logger


class MarketDataMain:
    def __init__(self):
        self.market_data = MarketData(
            api_key=API_KEY,
            secret_key=SECRET_KEY,
            passphrase=PASSPHRASE,
            sandbox=SANDBOX,
        )
        self.symbols = MONITOR_CONFIG.get("volume_monitor", {}).get(
            "symbols", ["XCH-USDT"]
        )
        self.bars = MONITOR_CONFIG.get("volume_monitor", {}).get(
            "bars", ["5m", "15m", "1H", "1D"]
        )
        self.initial_date = MONITOR_CONFIG.get("volume_monitor", {}).get(
            "initial_date", "2025-07-01 00:00:00"
        )
        mysql_user = MYSQL_CONFIG.get("user", "xch")
        mysql_password = MYSQL_CONFIG.get("password", "")
        if not mysql_password:
            raise ValueError("MySQL password is not set")
        mysql_host = MYSQL_CONFIG.get("host", "localhost")
        mysql_port = MYSQL_CONFIG.get("port", 3306)
        mysql_database = MYSQL_CONFIG.get("database", "okx")

        self.db_url = f"mysql+pymysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_database}"
        self.db_market_data = DBMarketData(self.db_url)
        self.trade_data_main = TradeDataMain()

    def initial_data(self):
        """
        初始化数据
        """
        for symbol in self.symbols:
            for bar in self.bars:
                logger.info(f"开始初始化行情数据: {symbol} {bar}")
                latest_data = self.db_market_data.query_latest_data(symbol, bar)
                if latest_data:
                    start = latest_data.get("timestamp")
                    start_date_time = timestamp_to_datetime(start)
                    start = start + 1
                else:
                    start = datetime_to_timestamp(self.initial_date)
                    start_date_time = self.initial_date
                logger.info(
                    f"开始初始化{symbol}, {bar} 行情数据，从 {start_date_time} 开始"
                )
                self.fetch_save_data(symbol, bar, start)

    def fetch_save_data(self, symbol: str, bar: str, start: str):
        """
        获取保存数据
        """
        end_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        end_time_ts = transform_date_time_to_timestamp(end_time)
        if end_time_ts is None:
            logger.error(f"结束时间格式错误: {end_time}")
            return None

        start_time_ts = transform_date_time_to_timestamp(start)
        if start_time_ts is None:
            logger.error(f"开始时间格式错误: {start}")
            return None

        # 如果bar为5m, 15m, 30m:
        # end_time_ts与start_time_ts相差超过1天，则按照1天为单位
        # 如果bar为1H, 4H,
        # end_time_ts与start_time_ts相差超过5天，则按照5天为单位
        # 如果bar为1D, 则end_time_ts与start_time_ts相差超过10天，则按照10天为单位
        # 获取数据，直到end_time_ts
        threshold = None
        if bar in ["5m", "15m", "30m"]:
            threshold = 86400000
        elif bar in ["1H", "4H"]:
            threshold = 432000000
        elif bar == "1D":
            threshold = 864000000

        get_data = False
        min_start_time_ts = start_time_ts
        while start_time_ts < end_time_ts:
            current_start_time_ts = end_time_ts - threshold
            if current_start_time_ts < start_time_ts:
                current_start_time_ts = start_time_ts
            start_date_time = timestamp_to_datetime(current_start_time_ts)
            end_date_time = timestamp_to_datetime(end_time_ts)
            logger.info(
                f"获取行情数据: {symbol} {bar} 从 {start_date_time} 到 {end_date_time}"
            )
            data = self.market_data.get_historical_kline_data(
                symbol=symbol,
                start=current_start_time_ts,
                bar=bar,
                end_time=end_time_ts,
            )
            if data is not None and len(data) > 0:
                data["buy_sz"] = -1
                data["sell_sz"] = -1

            # 根据交易数据，设置buy_sz和sell_sz
            # 比特币的数据获取过慢，暂时不获取交易数据
            # if not symbol.endswith("-SWAP"):
            #     # trade_data的end_time需要比market_data的end_time大一个周期
            #     trade_data = self.trade_data_main.get_trade_data(
            #         symbol=symbol, start_time=current_start_time_ts, end_time=end_time_ts
            #     )
            #     for index, row in data.iterrows():
            #         try:
            #             current_from_time = int(row["timestamp"])
            #             if index == len(data) - 1:
            #                 current_to_time = current_from_time + BAR_THRESHOLD[bar]
            #             else:
            #                 current_to_time = int(data.iloc[index + 1]["timestamp"])
            #             current_trade_data = trade_data[
            #                 (trade_data["ts"] >= current_from_time)
            #                 & (trade_data["ts"] <= current_to_time)
            #             ]
            #             if current_trade_data is not None and len(current_trade_data) > 0:
            #                 current_buy_sz = current_trade_data[
            #                     current_trade_data["side"] == "buy"
            #                 ]["sz"].sum()
            #                 current_sell_sz = current_trade_data[
            #                     current_trade_data["side"] == "sell"
            #                 ]["sz"].sum()
            #                 data.loc[index, "buy_sz"] = current_buy_sz
            #                 data.loc[index, "sell_sz"] = current_sell_sz
            #         except Exception as e:
            #             logger.error(f"设置buy_sz和sell_sz失败: {e}")
            #             continue
            if data is not None and len(data) > 0:
                data = data[
                    [
                        "symbol",
                        "bar",
                        "timestamp",
                        "date_time",
                        "open",
                        "high",
                        "low",
                        "close",
                        "volume",
                        "volCcy",
                        "volCCyQuote",
                        "buy_sz",
                        "sell_sz",
                        "create_time",
                    ]
                ]
                data = self.add_new_columns(data)
                self.db_market_data.insert_data_to_mysql(data)
                current_min_start_time_ts = data["timestamp"].min()
                if current_min_start_time_ts < min_start_time_ts:
                    min_start_time_ts = current_min_start_time_ts
                get_data = True
            if current_start_time_ts == start_time_ts:
                break
            end_time_ts = current_start_time_ts
        if min_start_time_ts is not None and get_data:
            # 补充技术指标数据
            # 获得min_start_time_ts之前30条数据
            logger.info(f"开始补充技术指标数据: {symbol} {bar}")
            before_data = self.db_market_data.query_data_before_timestamp(
                symbol, bar, min_start_time_ts, 30
            )
            latest_before_timestamp = None
            if before_data is not None and len(before_data) > 0:
                earliest_timestamp = before_data[-1]["timestamp"]
                latest_before_timestamp = before_data[0]["timestamp"]
            else:
                earliest_timestamp = min_start_time_ts
            handle_data = self.db_market_data.query_market_data_by_symbol_bar(
                symbol=symbol, bar=bar, start=earliest_timestamp, end=None
            )
            if handle_data is not None:
                if before_data is not None and len(handle_data) <= len(before_data):
                    logger.error(f"handle_data数据条数小于before_data数据条数: {symbol} {bar}")
                    return None
                if isinstance(handle_data, list):
                    handle_data = pd.DataFrame(handle_data)
                elif isinstance(handle_data, dict):
                    handle_data = pd.DataFrame([handle_data])
                elif isinstance(handle_data, pd.DataFrame):
                    pass
                else:
                    logger.error(f"handle_data类型错误: {type(handle_data)}")
                    return None

                handle_data = self.calculate_metrics(handle_data)
                if latest_before_timestamp is not None:
                    handle_data = handle_data[handle_data["timestamp"] > latest_before_timestamp]
                    handle_data.reset_index(drop=True, inplace=True)
                logger.info(f"开始保存技术指标数据: {symbol} {bar}")
                self.db_market_data.insert_data_to_mysql(handle_data)
        return data
    
    def add_new_columns(self, data: pd.DataFrame):
        """
        添加新列
        """
        columns = data.columns.tolist()
        if "buy_sz" not in columns:
            data["buy_sz"] = -1
        if "sell_sz" not in columns:
            data["sell_sz"] = -1
        data["pre_close"] = None
        data["close_change"] = None
        data["pct_chg"] = None
        data["ma1"] = None
        data["ma2"] = None
        data["dif"] = None
        data["dea"] = None
        data["macd"] = None
        data["macd_signal"] = None
        data["macd_divergence"] = None
        data["kdj_k"] = None
        data["kdj_d"] = None
        data["kdj_j"] = None
        data["kdj_signal"] = None
        data["kdj_pattern"] = None
        data["ma5"] = None
        data["ma10"] = None
        data["ma20"] = None
        data["ma30"] = None
        data["ma_cross"] = None
        data["ma5_close_diff"] = None
        data["ma10_close_diff"] = None
        data["ma20_close_diff"] = None
        data["ma30_close_diff"] = None
        data["ma_close_avg"] = None
        data["ma_long_short"] = None
        data["ma_divergence"] = None
        data["rsi_14"] = None
        data["rsi_signal"] = None
        data["boll_upper"] = None
        data["boll_middle"] = None
        data["boll_lower"] = None
        data["boll_signal"] = None
        data["boll_pattern"] = None
        data["k_length"] = None
        data["k_shape"] = None
        data["k_up_down"] = None
        return data

    def calculate_metrics(self, data: pd.DataFrame):
        """
        计算技术指标
        1. 计算前一日收盘价、涨跌幅、涨跌幅百分比
        2. 计算MACD指标
        3. 计算KDJ指标
        4. 计算BOLL指标
        5. 计算K线长度
        6. 计算K线形状
        7. 计算K线方向
        pre_close DECIMAL(20,10) NULL,
        close_change DECIMAL(20,10) NULL,
        pct_chg DECIMAL(20,10)   NULL,
        ma1 DOUBLE DEFAULT NULL COMMENT '移动平均线1',
        ma2 DOUBLE DEFAULT NULL COMMENT '移动平均线2',
        dif DOUBLE DEFAULT NULL COMMENT 'MACD指标DIF线',
        dea DOUBLE DEFAULT NULL COMMENT 'MACD指标DEA线',
        macd DOUBLE DEFAULT NULL COMMENT 'MACD指标值',
        macd_signal VARCHAR(15) DEFAULT NULL COMMENT 'MACD金叉死叉信号',
        macd_divergence varchar(25) DEFAULT NULL COMMENT 'MACD背离，顶背离或底背离',
        kdj_k DOUBLE DEFAULT NULL COMMENT 'KDJ指标K值',
        kdj_d DOUBLE DEFAULT NULL COMMENT 'KDJ指标D值',
        kdj_j DOUBLE DEFAULT NULL COMMENT 'KDJ指标J值',
        kdj_signal VARCHAR(15) DEFAULT NULL COMMENT 'KDJ金叉死叉信号',
        kdj_pattern varchar(25) DEFAULT NULL COMMENT 'KDJ超买，超卖，徘徊',
        ma5 DOUBLE DEFAULT NULL COMMENT '5移动平均线',
        ma10 DOUBLE DEFAULT NULL COMMENT '10移动平均线',
        ma20 DOUBLE DEFAULT NULL COMMENT '20移动平均线',
        ma30 DOUBLE DEFAULT NULL COMMENT '30移动平均线',
        ma_cross VARCHAR(15) DEFAULT NULL COMMENT '均线交叉信号',
        ma5_close_diff double DEFAULT NULL COMMENT '5移动平均线与收盘价差值',
        ma10_close_diff double DEFAULT NULL COMMENT '10移动平均线与收盘价差值',
        ma20_close_diff double DEFAULT NULL COMMENT '20移动平均线与收盘价差值',
        ma30_close_diff double DEFAULT NULL COMMENT '30移动平均线与收盘价差值',
        ma_close_avg double DEFAULT NULL COMMENT '收盘价移动平均值',
        ma_long_short varchar(25) DEFAULT NULL COMMENT '均线多空',
        ma_divergence varchar(25) DEFAULT NULL COMMENT '均线发散，均线粘合，均线适中，均线发散，均线超发散'
        rsi_14 DOUBLE DEFAULT NULL COMMENT '14RSI指标',
        rsi_signal VARCHAR(15) DEFAULT NULL COMMENT 'RSI强弱信号',
        boll_upper DOUBLE DEFAULT NULL COMMENT '布林带上轨',
        boll_middle DOUBLE DEFAULT NULL COMMENT '布林带中轨',
        boll_lower DOUBLE DEFAULT NULL COMMENT '布林带下轨',
        boll_signal VARCHAR(15) DEFAULT NULL COMMENT '布林带强弱信号',
        boll_pattern varchar(25) DEFAULT NULL COMMENT 'BOLL超买，超卖，徘徊',
        k_length varchar(25) DEFAULT NULL COMMENT 'K线长度',
        k_shape varchar(25) DEFAULT NULL COMMENT 'K线形状',
        k_up_down varchar(25) DEFAULT NULL COMMENT 'K线方向',
        """
        data = data.sort_values(by="timestamp")
        data = data.reset_index(drop=True)

        metrics_calculation = MetricsCalculation()
        data = metrics_calculation.pre_close(data)
        data = metrics_calculation.macd(data)
        data = metrics_calculation.kdj(data)
        data = metrics_calculation.set_kdj_pattern(data)
        data = metrics_calculation.update_macd_divergence_column_simple(data)
        data = metrics_calculation.ma5102030(data)
        data = metrics_calculation.calculate_ma_price_percent(data)
        data = metrics_calculation.set_ma_long_short_divergence(data)
        data = metrics_calculation.rsi(data)
        data = metrics_calculation.boll(data)
        data = metrics_calculation.set_boll_pattern(data)
        data = metrics_calculation.set_k_length(data)
        data = metrics_calculation.set_k_shape(data)
        return data

    def batch_update_data(self):
        """
        更新数据
        1. 获取最新数据
        2. 获取最新数据的时间戳
        3. 根据最新数据的时间戳，获取最新数据
        4. 将最新数据保存到数据库
        """
        for symbol in self.symbols:
            for bar in self.bars:
                self.update_data(symbol, bar)

    def update_data(self, symbol: str, bar: str):
        """
        更新数据
        """
        logger.info(f"开始更新行情数据: {symbol} {bar}")
        latest_data = self.db_market_data.query_latest_data(symbol, bar)
        if not latest_data:
            logger.info(f"{symbol}, {bar} 无数据，开始从{self.initial_date}初始化数据")
            data = self.fetch_save_data(symbol, bar, self.initial_date)
        else:
            latest_timestamp = latest_data.get("timestamp")
            if latest_timestamp:
                latest_timestamp = int(latest_timestamp)
                latest_date_time = timestamp_to_datetime(latest_timestamp)
                logger.info(
                    f"{symbol}, {bar} 上次获取的最新数据时间: {latest_date_time}"
                )
            else:
                logger.warning(f"获取{symbol}, {bar} 最新数据失败")
                return
            data = self.fetch_save_data(symbol, bar, latest_timestamp + 1)
        return data
    
    def batch_calculate_metrics(self):
        """
        批量计算技术指标
        """
        logger.info("开始批量计算技术指标")
        start_date_time = MONITOR_CONFIG.get("volume_monitor", {}).get(
            "initial_date", "2025-05-15 00:00:00"
        )
        start_timestamp = transform_date_time_to_timestamp(start_date_time)
        current_date_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        current_timestamp = transform_date_time_to_timestamp(current_date_time)
        for symbol in self.symbols:
            for bar in self.bars:
                logger.info(f"开始计算技术指标: {symbol} {bar}")
                data = self.db_market_data.query_market_data_by_symbol_bar(
                    symbol=symbol, bar=bar, start=start_timestamp - 1, end=current_timestamp
                )
                if data is not None and len(data) > 0:
                    data = pd.DataFrame(data)
                    data = self.calculate_metrics(data)
                    logger.info(f"开始保存技术指标数据: {symbol} {bar}")
                    self.db_market_data.insert_data_to_mysql(data)


if __name__ == "__main__":
    market_data_main = MarketDataMain()
    # market_data_main.batch_update_data()
    # market_data_main.initial_data()
    market_data_main.batch_calculate_metrics()