265 lines
11 KiB
Python
265 lines
11 KiB
Python
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=100,
|
||
)
|
||
|
||
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_chg"] > 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,
|
||
real_time_data,
|
||
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:
|
||
if self.latest_record.get(symbol, {}).get(bar, 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=100
|
||
)
|
||
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="PUMP-USDT",
|
||
bar="15m",
|
||
only_output_huge_volume=False,
|
||
only_output_rise=False,
|
||
)
|