support get market data by API

support store data to mysql database.

config.py

@@ -50,7 +50,8 @@ TIME_CONFIG = {
 
 MONITOR_CONFIG = {
     "volume_monitor":{
-        "symbols": ["XCH-USDT"],
+        "symbols": ["XCH-USDT", "BTC-USDT", "ETH-USDT", "SOL-USDT", "DOGE-USDT", 
+        "XCH-USDT-SWAP", "BTC-USDT-SWAP", "ETH-USDT-SWAP", "SOL-USDT-SWAP", "DOGE-USDT-SWAP"],
         "intervals": ["5m", "15m", "1H", "4H", "1D"],
         "initial_date": "2025-07-01 00:00:00"
     },
@@ -62,4 +63,12 @@ MONITOR_CONFIG = {
             {"interval": "1H", "threshold": 0.1}
             ]
     }
+}
+
+MYSQL_CONFIG = {
+    "host": "localhost",
+    "port": 3306,
+    "user": "xch",
+    "password": "xch_okx_2025",
+    "database": "okx"
 }

core/data_monitor.py

@@ -41,8 +41,11 @@ class DataMonitor:
             start_time = int(two_months_ago.timestamp() * 1000)
         else:
             try:
+                # 判断是否就是timestamp整型数据
+                if isinstance(start, int):
+                    start_time = start
                 # 判断是否为纯数字（UTC毫秒级timestamp）
-                if start.isdigit():
+                elif start.isdigit():
                     start_time = int(start)
                 else:
                     # 按北京时间字符串处理，转换为毫秒级timestamp
@@ -52,6 +55,7 @@ class DataMonitor:
                 return None
         columns = ["timestamp", "open", "high", "low", "close", "volume", "volCcy", "volCCyQuote", "confirm"]
         all_data = []
+        latest_timestamp = -1
         while start_time < end_time:
             try:
                 # after，真实逻辑是获得指定时间之前的数据 ！！！
@@ -65,23 +69,50 @@ class DataMonitor:
                     logging.warning(f"请求失败或无数据: {response.get('msg', 'No message')}")
                     break
                 candles = response["data"]
-                all_data.extend(candles)
+
-                # 更新 end_time 为本次请求中最早的时间戳
+                from_time = int(candles[-1][0])
-                end_time = int(candles[-1][0])
+                to_time = int(candles[0][0])
-                logging.info(f"已获取 {len(candles)} 条数据，最早时间: {pd.to_datetime(end_time, unit='ms', utc=True).tz_convert('Asia/Shanghai')}")
+                if latest_timestamp == -1:
-                end_time -= 1  # 减 1 毫秒以避免重复
+                    latest_timestamp = from_time
-                time.sleep(0.2)
+                else:
+                    if latest_timestamp > from_time:
+                        latest_timestamp = from_time
+                    else:
+                        logging.warning(f"上一次数据最早时间戳 {latest_timestamp} 小于等于 from_time {from_time}， 停止获取数据")
+                        break
+                from_time_str = pd.to_datetime(from_time, unit='ms', utc=True).tz_convert('Asia/Shanghai')
+                to_time_str = pd.to_datetime(to_time, unit='ms', utc=True).tz_convert('Asia/Shanghai')
+                logging.info(f"已获取 {len(candles)} 条数据，从: {from_time_str} 到: {to_time_str}")
+                if from_time < start_time:
+                    start_time_str = pd.to_datetime(start_time, unit='ms', utc=True).tz_convert('Asia/Shanghai')
+                    logging.warning(f"本轮获取数据最早时间 {from_time_str} 早于 此次数据获取起始时间 {start_time_str}， 停止获取数据")
+                    # candels中仅保留start_time之后的数据
+                    candles = [candle for candle in candles if int(candle[0]) >= start_time]
+                    all_data.extend(candles)
+                    break
+                else:
+                    # 更新 end_time 为本次请求中最早的时间戳
+                    end_time = from_time - 1  # 减 1 毫秒以避免重复
+                    all_data.extend(candles)
+                    time.sleep(0.2)
             except Exception as e:
                 logging.error(f"请求出错: {e}")
                 break
+
         if all_data:
             df = pd.DataFrame(all_data, columns=columns)
             df = df[df['confirm'] == '1']
-            for col in ['open', 'high', 'low', 'close', 'volume']:
+            for col in ['open', 'high', 'low', 'close', 'volume', 'volCcy', 'volCCyQuote']:
                 df[col] = pd.to_numeric(df[col], errors='coerce')
             dt_series = pd.to_datetime(df['timestamp'].astype(int), unit='ms', utc=True, errors='coerce').dt.tz_convert('Asia/Shanghai')
             df['date_time'] = dt_series.dt.strftime('%Y-%m-%d %H:%M:%S')
-            df = df[['timestamp', 'date_time', 'open', 'high', 'low', 'close', 'volume']]
+            # 将timestamp转换为整型
+            df['timestamp'] = df['timestamp'].astype(int)
+            # 添加虚拟货币名称列，内容为symbol
+            df['symbol'] = symbol
+            # 添加bar列，内容为bar
+            df['bar'] = bar
+            df = df[['symbol', 'bar', 'timestamp', 'date_time', 'open', 'high', 'low', 'close', 'volume', 'volCcy', 'volCCyQuote']]
             df.sort_values('timestamp', inplace=True)
             df.reset_index(drop=True, inplace=True)
             logging.info(f"总计获取 {len(df)} 条 K 线数据（仅confirm=1）")

core/db_manager.py

@@ -0,0 +1,56 @@
+import pandas as pd
+from sqlalchemy import create_engine, exc, text
+import logging
+
+def save_market_data_to_mysql(df: pd.DataFrame, db_url: str):
+    """
+    将K线行情数据保存到MySQL的crypto_market_data表
+    :param df: K线数据DataFrame
+    :param symbol: 交易对
+    :param bar: K线周期
+    :param db_url: 数据库连接URL
+    """
+    if df is None or df.empty:
+        logging.warning("DataFrame为空，无需写入数据库。")
+        return
+
+    # 按表字段顺序排列
+    columns = [
+        'symbol', 'bar', 'timestamp', 'date_time', 'open', 'high', 'low', 'close',
+        'volume', 'volCcy', 'volCCyQuote'
+    ]
+    df = df[columns]
+
+    # 建立数据库连接
+    try:
+        engine = create_engine(db_url)
+        try:
+            df.to_sql(
+                name='crypto_market_data',
+                con=engine,
+                if_exists='append',
+                index=False,
+                method='multi'
+            )
+            logging.info("数据已成功写入数据库。")
+        except Exception as e:
+            logging.error(f'插入数据出错: {e}')
+            with engine.begin() as conn:
+                for _, row in df.iterrows():
+                    try:
+                        sql = text("""
+                            INSERT INTO crypto_market_data
+                            (symbol, bar, timestamp, date_time, open, high, low, close, volume, volCcy, volCCyQuote)
+                            VALUES (:symbol, :bar, :timestamp, :date_time, :open, :high, :low, :close, :volume, :volCcy, :volCCyQuote)
+                            ON DUPLICATE KEY UPDATE
+                                open=VALUES(open), high=VALUES(high), low=VALUES(low), close=VALUES(close),
+                                volume=VALUES(volume), volCcy=VALUES(volCcy), volCCyQuote=VALUES(volCCyQuote), date_time=VALUES(date_time)
+                        """)
+                        conn.execute(sql, row.to_dict())
+                    except exc.IntegrityError as e:
+                        logging.error(f'唯一索引冲突: {e}')
+                    except Exception as e:
+                        logging.error(f'插入数据出错: {e}')
+                logging.info("数据已成功写入数据库。")
+    except Exception as e:
+        logging.error(f'数据库连接或写入失败: {e}')

monitor_main.py

@@ -1,11 +1,12 @@
 import logging
 from time import sleep
 from core.data_monitor import DataMonitor
-from config import API_KEY, SECRET_KEY, PASSPHRASE, SANDBOX, MONITOR_CONFIG
+from core.db_manager import save_market_data_to_mysql
+from config import API_KEY, SECRET_KEY, PASSPHRASE, SANDBOX, \
+    MONITOR_CONFIG, MYSQL_CONFIG
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s')
 
-
 class MonitorMain:
     def __init__(self):
         self.data_monitor = DataMonitor(
@@ -17,6 +18,15 @@ class MonitorMain:
         self.symbols = MONITOR_CONFIG.get("volume_monitor", {}).get("symbols", ["XCH-USDT"])
         self.intervals = MONITOR_CONFIG.get("volume_monitor", {}).get("intervals", ["5m", "15m", "1H", "4H", "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}"
     
     def initial_data(self):
         for symbol in self.symbols:
@@ -24,6 +34,7 @@ class MonitorMain:
                 data = self.data_monitor.get_historical_kline_data(symbol=symbol, 
                 start=self.initial_date, 
                 bar=interval)
+                save_market_data_to_mysql(data, self.db_url)
     
 
 if __name__ == "__main__":

sql/table/crypto_market_data.sql

@@ -0,0 +1,15 @@
+CREATE TABLE IF NOT EXISTS crypto_market_data (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY,
+    symbol VARCHAR(50) NOT NULL,
+    bar VARCHAR(20) NOT NULL,
+    timestamp BIGINT NOT NULL,
+    date_time VARCHAR(50) NOT NULL,
+    open DECIMAL(20,5) NOT NULL,
+    high DECIMAL(20,5) NOT NULL,
+    low DECIMAL(20,5) NOT NULL,
+    close DECIMAL(20,5) NOT NULL,
+    volume DECIMAL(30,8) NOT NULL,
+    volCcy DECIMAL(30,8) NOT NULL,
+    volCCyQuote DECIMAL(30,8) NOT NULL,
+    UNIQUE KEY uniq_symbol_bar_timestamp (symbol, bar, timestamp)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;