crypto_quant/market_monitor_main.py

from numpy import real
from market_data_main import MarketDataMain
from huge_volume_main import HugeVolumeMain
from core.biz.market_monitor import create_metrics_report
from core.db.db_market_monitor import DBMarketMonitor
from core.wechat import Wechat
from config import MONITOR_CONFIG, MYSQL_CONFIG
from core.utils import timestamp_to_datetime, transform_date_time_to_timestamp

import logging
import os
import pandas as pd
from datetime import datetime, timedelta
import json
import re

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")


class MarketMonitorMain:
    def __init__(self):
        self.market_data_main = MarketDataMain()
        self.huge_volume_main = HugeVolumeMain()
        self.wechat = Wechat()
        self.monitor_config = MONITOR_CONFIG
        self.window_size = 100
        self.start_date = MONITOR_CONFIG.get("volume_monitor", {}).get(
            "initial_date", "2025-05-01 00:00:00"
        )
        self.latest_record_file_path = "./output/record/latest_record.json"
        self.latest_record = self.get_latest_record()
        self.output_folder = "./output/report/market_monitor/"
        os.makedirs(self.output_folder, exist_ok=True)

        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_monitor = DBMarketMonitor(self.db_url)

    def get_latest_record(self):
        """
        获取最新记录
        """
        os.makedirs(os.path.dirname(self.latest_record_file_path), exist_ok=True)
        try:
            if os.path.exists(self.latest_record_file_path):
                with open(self.latest_record_file_path, "r", encoding="utf-8") as f:
                    return json.load(f)
            else:
                with open(self.latest_record_file_path, "w", encoding="utf-8") as f:
                    json.dump({}, f, ensure_ascii=False, indent=4)
                return {}
        except Exception as e:
            logging.error(f"获取最后一次报表生成记录失败: {e}")
            return {}

    def monitor_realtime_market(
        self,
        symbol: str,
        bar: str,
        only_output_huge_volume: bool = False,
        only_output_rise: bool = False,
    ):
        """
        监控最新市场数据
        考虑到速度，暂不与数据库交互，直接从api获取数据
        """
        # 获得当前时间字符串
        now_datetime = datetime.now()
        now_datetime_str = now_datetime.strftime("%Y-%m-%d %H:%M:%S")
        end_time = transform_date_time_to_timestamp(now_datetime_str)
        real_time_data = self.market_data_main.market_data.get_realtime_kline_data(
            symbol=symbol,
            bar=bar,
            end_time=end_time,
            limit=50,
        )

        if real_time_data is None or len(real_time_data) == 0:
            logging.error(f"获取最新市场数据失败: {symbol}, {bar}")
            return

        latest_realtime_timestamp = real_time_data["timestamp"].iloc[-1]
        latest_realtime_timestamp = int(latest_realtime_timestamp)
        latest_record_timestamp = (
            self.latest_record.get(symbol, {}).get(bar, {}).get("timestamp", None)
        )
        latest_reatime_datetime = timestamp_to_datetime(latest_realtime_timestamp)
        if latest_record_timestamp is not None:
            latest_record_timestamp = int(latest_record_timestamp)

            latest_record_datetime = timestamp_to_datetime(latest_record_timestamp)
            if (
                latest_record_timestamp is not None
                and latest_realtime_timestamp <= latest_record_timestamp
            ):
                logging.info(
                    f"最新市场数据时间戳 {latest_reatime_datetime} 小于等于最新记录时间戳 {latest_record_datetime}， 不进行监控"
                )
                return
            logging.info(
                f"最新市场数据时间 {latest_reatime_datetime}, 上一次记录时间 {latest_record_datetime}"
            )
        else:
            logging.info(
                f"最新市场数据时间 {latest_reatime_datetime}, 上一次记录时间为空"
            )

        real_time_data = self.market_data_main.add_new_columns(real_time_data)
        logging.info(f"开始计算技术指标: {symbol} {bar}")
        real_time_data = self.market_data_main.calculate_metrics(real_time_data)
        logging.info(f"开始计算大成交量: {symbol} {bar} 窗口大小: {self.window_size}")
        real_time_data = self.huge_volume_main.huge_volume.detect_huge_volume(
            data=real_time_data,
            window_size=self.window_size,
            threshold=self.huge_volume_main.threshold,
            check_price=True,
            only_output_huge_volume=False,
            output_excel=False,
        )
        if real_time_data is None or len(real_time_data) == 0:
            logging.error(
                f"计算大成交量失败: {symbol} {bar} 窗口大小: {self.window_size}"
            )
            return
        realtime_row = real_time_data.iloc[-1]

        if only_output_huge_volume:
            if realtime_row["huge_volume"] == 1:
                logging.info(f"监控到巨量: {symbol} {bar} 窗口大小: {self.window_size}")
            else:
                logging.info(
                    f"监控到非巨量: {symbol} {bar} 窗口大小: {self.window_size}，退出本次监控"
                )
                return
        if only_output_rise:
            if realtime_row["pct_change"] > 0:
                logging.info(f"监控到上涨: {symbol} {bar} 窗口大小: {self.window_size}")
            else:
                logging.info(
                    f"监控到下跌: {symbol} {bar} 窗口大小: {self.window_size}，退出本次监控"
                )
                return

        next_bar_row = self.get_other_realtime_data(symbol, bar, end_time, next=True)
        if "BTC-USDT" in symbol:
            btc_bar_row = None
        else:
            btc_bar_row = self.get_other_realtime_data(
                "BTC-USDT", bar, end_time, next=False
            )

        report = create_metrics_report(
            realtime_row,
            next_bar_row,
            btc_bar_row,
            only_output_huge_volume,
            only_output_rise,
            now_datetime_str,
        )
        text_length = len(report.encode("utf-8"))

        logging.info(f"发送报告到企业微信，字节数: {text_length}")
        self.wechat.send_markdown(report)

        # remove punction in latest_reatime_datetime
        file_datetime = re.sub(r"[\:\-\s]", "", latest_reatime_datetime)
        report_file_name = f"{symbol}_{bar}_{self.window_size}_{file_datetime}.md"
        report_file_path = os.path.join(self.output_folder, report_file_name)
        with open(report_file_path, "w", encoding="utf-8") as f:
            f.write(report)
        report_file_byte_size = os.path.getsize(report_file_path)
        report_data = {
            "symbol": symbol,
            "bar": bar,
            "window_size": self.window_size,
            "timestamp": latest_realtime_timestamp,
            "date_time": latest_reatime_datetime,
            "report": report,
            "report_file_path": report_file_path,
            "report_file_name": report_file_name,
            "report_file_byte_size": report_file_byte_size,
        }
        report_data = pd.DataFrame([report_data])
        logging.info(f"插入数据到数据库")
        self.db_market_monitor.insert_data_to_mysql(report_data)
        if self.latest_record.get(symbol, None) is None:
            self.latest_record[symbol] = {bar: {"timestamp": latest_realtime_timestamp}}
        else:
            self.latest_record[symbol][bar]["timestamp"] = latest_realtime_timestamp
        with open(self.latest_record_file_path, "w", encoding="utf-8") as f:
            json.dump(self.latest_record, f, ensure_ascii=False, indent=4)

    def get_other_realtime_data(
        self, symbol: str, bar: str, end_time: int, next: bool = True
    ):
        """
        获取下一个长周期实时数据
        """
        if next:
            # 获得bar在self.market_data_main.bars中的索引
            bar_index = self.market_data_main.bars.index(bar)
            if bar_index == len(self.market_data_main.bars) - 1:
                logging.error(f"已经是最后一个bar: {bar}")
                return None
            # 获得下一个bar
            bar = self.market_data_main.bars[bar_index + 1]
            # 获得下一个bar的实时数据
        data = self.market_data_main.market_data.get_realtime_kline_data(
            symbol=symbol, bar=bar, end_time=end_time, limit=50
        )
        if data is None or len(data) == 0:
            logging.error(f"获取实时数据失败: {symbol}, {bar}")
            return None
        data = self.market_data_main.add_new_columns(data)
        logging.info(f"开始计算技术指标: {symbol} {bar}")
        data = self.market_data_main.calculate_metrics(data)
        row = data.iloc[-1]
        return row

    def batch_monitor_realtime_market(
        self,
        only_output_huge_volume: bool = True,
        only_output_rise: bool = True,
    ):
        for symbol in self.market_data_main.symbols:
            for bar in self.market_data_main.bars:
                logging.info(
                    f"开始监控: {symbol} {bar} 窗口大小: {self.window_size} 行情数据"
                )
                try:
                    self.monitor_realtime_market(
                        symbol,
                        bar,
                        only_output_huge_volume,
                        only_output_rise,
                    )
                except Exception as e:
                    logging.error(
                        f"监控失败: {symbol} {bar} 窗口大小: {self.window_size} 行情数据: {e}"
                    )
                    continue


if __name__ == "__main__":
    market_monitor_main = MarketMonitorMain()
    market_monitor_main.monitor_realtime_market(
        symbol="ETH-USDT",
        bar="5m",
        only_output_huge_volume=True,
        only_output_rise=False,
    )