crypto_quant/market_data_main.py

import core.logger as logging
from datetime import datetime, timedelta, timezone
from core.utils import get_current_date_time
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.db.db_binance_data import DBBinanceData
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,
    OKX_MONITOR_CONFIG,
    BINANCE_MONITOR_CONFIG,
    US_STOCK_MONITOR_CONFIG,
    COIN_MYSQL_CONFIG,
    BAR_THRESHOLD,
)

logger = logging.logger


class MarketDataMain:
    def __init__(self, is_us_stock: bool = False, is_binance: bool = False):
        self.market_data = MarketData(
            api_key=API_KEY,
            secret_key=SECRET_KEY,
            passphrase=PASSPHRASE,
            sandbox=SANDBOX,
            is_us_stock=is_us_stock,
        )
        if is_us_stock:
            self.symbols = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "symbols", ["QQQ"]
            )
            self.bars = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "bars", ["5m"]
            )
            self.initial_date = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "initial_date", "2015-08-30 00:00:00"
            )
        elif is_binance:
            self.symbols = BINANCE_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "symbols", ["BTC-USDT"]
            )
            self.bars = BINANCE_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "bars", ["5m", "30m", "1H"]
            )
            self.initial_date = BINANCE_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "initial_date", "2017-08-17 00:00:00"
            )
        else:
            self.symbols = OKX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "symbols", ["XCH-USDT"]
            )
            self.bars = OKX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "bars", ["1m", "5m", "15m", "1H", "1D"]
            )
            self.initial_date = OKX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "initial_date", "2025-07-01 00:00:00"
            )
        mysql_user = COIN_MYSQL_CONFIG.get("user", "xch")
        mysql_password = COIN_MYSQL_CONFIG.get("password", "")
        if not mysql_password:
            raise ValueError("MySQL password is not set")
        mysql_host = COIN_MYSQL_CONFIG.get("host", "localhost")
        mysql_port = COIN_MYSQL_CONFIG.get("port", 3306)
        mysql_database = COIN_MYSQL_CONFIG.get("database", "okx")

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

    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 = get_current_date_time()
        end_time_ts = transform_date_time_to_timestamp(end_time)
        if end_time_ts is None:
            logger.error(f"结束时间格式错误: {end_time}")
            return None

        if bar == "1m":
            if isinstance(start, str):
                start_time_ts = transform_date_time_to_timestamp(start)
                # start为end_time_ts的前十五天
                if end_time_ts - start_time_ts > 86400000 * 15:
                    start_time_ts = end_time_ts - 86400000 * 15
                    start = timestamp_to_datetime(start_time_ts)
            else:
                start_time_ts = transform_date_time_to_timestamp(start)
        else:
            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 ["1m", "5m", "15m", "30m", "1H"]:
            if self.is_us_stock:
                if bar == "1m":
                    threshold = 86400000 * 2
                if bar == "5m":
                    threshold = 86400000 * 4
                elif bar == "15m":
                    threshold = 86400000 * 6
                elif bar == "30m":
                    threshold = 86400000 * 12
                elif bar == "1H":
                    threshold = 86400000 * 24
            else:
                threshold = 86400000
        elif bar in ["1H", "4H"]:
            threshold = 432000000
        elif bar == "1D":
            threshold = 864000000

        get_data = False
        min_start_time_ts = start_time_ts
        max_start_time_ts = None
        while start_time_ts < end_time_ts:
            current_start_time_ts = int(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}"
            )
            if self.is_us_stock:
                limit = 1000
            else:
                limit = 100
            data = self.market_data.get_historical_kline_data(
                symbol=symbol,
                start=current_start_time_ts,
                bar=bar,
                end_time=end_time_ts,
                limit=limit,
            )
            if data is not None and len(data) > 0:
                data = self.post_save_data(data)
                current_min_start_time_ts = int(data["timestamp"].min())
                if current_min_start_time_ts < min_start_time_ts:
                    min_start_time_ts = current_min_start_time_ts
                current_max_start_time_ts = int(data["timestamp"].max())
                if max_start_time_ts is None:
                    max_start_time_ts = current_max_start_time_ts
                else:
                    if current_max_start_time_ts > max_start_time_ts:
                        max_start_time_ts = current_max_start_time_ts
                get_data = True
            else:
                logger.warning(
                    f"获取行情数据为空: {symbol} {bar} 从 {start_date_time} 到 {end_date_time}"
                )
                break

            if current_start_time_ts == start_time_ts:
                break

            if current_min_start_time_ts < current_start_time_ts:
                end_time_ts = current_min_start_time_ts
            else:
                end_time_ts = current_start_time_ts
        if get_data:
            # 补充技术指标数据
            # 获得min_start_time_ts之前30条数据
            logger.info(f"开始补充技术指标数据: {symbol} {bar}")
            data = self.post_calculate_metrics(
                symbol, bar, min_start_time_ts, max_start_time_ts
            )
        return data
    
    def adjust_binance_csv_data(self, symbol: str, bar: str, data: pd.DataFrame):
        """
        调整binance csv数据
        """
        data["symbol"] = symbol
        data["bar"] = bar
        data["timestamp"] = None
        data["date_time"] = None
        data["date_time_us"] = None
        data["volCcy"] = None
        data["volCCyQuote"] = None
        data["create_time"] = None

        data = self.check_date_time(data, bar)

        for index, row in data.iterrows():
            candle_begin_time = row["candle_begin_time"]
            timestamp = datetime_to_timestamp(candle_begin_time, is_utc=True)
            data.loc[index, "timestamp"] = timestamp
            data.loc[index, "volCcy"] = row["quote_volume"]
            data.loc[index, "volCCyQuote"] = row["quote_volume"]
        data["timestamp"] = data["timestamp"].astype(int)
        dt_series = pd.to_datetime(data['timestamp'].astype(int), unit='ms', utc=True, errors='coerce').dt.tz_convert('Asia/Shanghai')
        data['date_time'] = dt_series.dt.strftime('%Y-%m-%d %H:%M:%S')
        dt_us_series = pd.to_datetime(data['timestamp'].astype(int), unit='ms', utc=True, errors='coerce').dt.tz_convert('America/New_York')
        data['date_time_us'] = dt_us_series.dt.strftime('%Y-%m-%d %H:%M:%S')
        data['create_time'] = get_current_date_time()

        data["date_time"] = data["date_time"].astype(str)
        data["date_time_us"] = data["date_time_us"].astype(str)
        data["open"] = data["open"].astype(float)
        data["high"] = data["high"].astype(float)
        data["low"] = data["low"].astype(float)
        data["close"] = data["close"].astype(float)
        data["volume"] = data["volume"].astype(float)
        data["volCcy"] = data["volCcy"].astype(float)
        data["volCCyQuote"] = data["volCCyQuote"].astype(float)
        data["create_time"] = data["create_time"].astype(str)
        data = data[
            [
                "symbol",
                "bar",
                "timestamp",
                "date_time",
                "date_time_us",
                "open",
                "high",
                "low",
                "close",
                "volume",
                "volCcy",
                "volCCyQuote",
                "create_time",
            ]
        ]
        data = data.sort_values(by="timestamp", ascending=True)
        data = data.reset_index(drop=True)
        return data
    
    def check_date_time(self, data: pd.DataFrame, bar: str):
        """
        检查日期时间
        """
        sample_date_time = data["candle_begin_time"].iloc[0]
        is_ok = True
        try:
            timestamp = datetime_to_timestamp(sample_date_time, is_utc=True)
        except Exception as e:
            is_ok = False
        if not is_ok:
            date_part = sample_date_time.split(" ")[0]
            first_date_time = f"{date_part} 00:00:00"
            first_date_time_utc = datetime.strptime(first_date_time, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
            if bar == "1H":
                # 将candle_begin_time以first_date_time为起点，每条记录增加一小时
                for index, row in data.iterrows():
                    if index == 0:
                        candle_begin_time = first_date_time_utc
                    else:
                        candle_begin_time = first_date_time_utc + timedelta(hours=1 * index)
                    data.loc[index, "candle_begin_time"] = candle_begin_time.strftime("%Y-%m-%d %H:%M:%S")
            elif bar == "5m":
                # 将candle_begin_time以first_date_time为起点，每条记录增加五分钟
                for index, row in data.iterrows():
                    if index == 0:
                        candle_begin_time = first_date_time_utc
                    else:
                        candle_begin_time = first_date_time_utc + timedelta(minutes=5 * index)
                    data.loc[index, "candle_begin_time"] = candle_begin_time.strftime("%Y-%m-%d %H:%M:%S")
            elif bar == "30m":
                # 将candle_begin_time以first_date_time为起点，每条记录增加三十分钟
                for index, row in data.iterrows():
                    if index == 0:
                        candle_begin_time = first_date_time_utc
                    else:
                        candle_begin_time = first_date_time_utc + timedelta(minutes=30 * index)
                    data.loc[index, "candle_begin_time"] = candle_begin_time.strftime("%Y-%m-%d %H:%M:%S")
            else:
                pass
        return data

    def post_save_data(self, data: pd.DataFrame):
        if data is not None and len(data) > 0:
            data["buy_sz"] = -1
            data["sell_sz"] = -1

            data = data[
                [
                    "symbol",
                    "bar",
                    "timestamp",
                    "date_time",
                    "date_time_us",
                    "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)
        return data

    def post_calculate_metrics(
        self, symbol: str, bar: str, min_start_time_ts: int, max_start_time_ts: int
    ):
        logger.info(f"开始补充技术指标数据: {symbol} {bar}")
        before_data = self.db_market_data.query_data_before_timestamp(
            symbol, bar, min_start_time_ts, 31
        )
        if before_data is not None and len(before_data) > 0:
            earliest_timestamp = before_data[-1]["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=max_start_time_ts
        )
        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)
            handle_data = handle_data[handle_data["timestamp"] >= min_start_time_ts]
            handle_data.reset_index(drop=True, inplace=True)
            logger.info(f"开始保存技术指标数据: {symbol} {bar}")
            self.db_market_data.insert_data_to_mysql(handle_data)
        return handle_data

    def add_new_columns(self, data: pd.DataFrame):
        """
        添加新列
        """
        data = data.copy()
        columns = data.columns.tolist()
        if "buy_sz" not in columns:
            data.loc[:, "buy_sz"] = -1
        if "sell_sz" not in columns:
            data.loc[:, "sell_sz"] = -1

        new_cols = [
            "pre_close",
            "close_change",
            "pct_chg",
            "ma1",
            "ma2",
            "dif",
            "dea",
            "macd",
            "macd_signal",
            "macd_divergence",
            "kdj_k",
            "kdj_d",
            "kdj_j",
            "kdj_signal",
            "kdj_pattern",
            "sar",
            "sar_signal",
            "ma5",
            "ma10",
            "ma20",
            "ma30",
            "ma_cross",
            "ma5_close_diff",
            "ma10_close_diff",
            "ma20_close_diff",
            "ma30_close_diff",
            "ma_close_avg",
            "ma_long_short",
            "ma_divergence",
            "rsi_14",
            "rsi_signal",
            "boll_upper",
            "boll_middle",
            "boll_lower",
            "boll_signal",
            "boll_pattern",
            "k_length",
            "k_shape",
            "k_up_down",
        ]
        for col in new_cols:
            data.loc[:, col] = pd.NA
        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超买，超卖，徘徊',
        sar DOUBLE DEFAULT NULL COMMENT 'SAR指标',
        sar_signal VARCHAR(15) DEFAULT NULL COMMENT 'SAR多头，SAR空头，SAR观望',
        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.sar(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 = OKX_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)

    def batch_ma_break_statistics(self):
        """
        批量计算MA突破统计
        """
        logger.info("开始批量计算MA突破统计")
        self.ma_break_statistics.batch_statistics(all_change=False)
        self.ma_break_statistics.batch_statistics(all_change=True)


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