feat(frontend): 添加交易看板,统一时区为CST
refactor(db): 移除所有SQLite相关代码,迁移至MySQL
This commit is contained in:
1 parent
5e0d6b18ef
commit
55453d3739
12 files changed
+2116
-501
No files matched your search
+77
-121
@@ -1,200 +1,156 @@
|
||||
import sqlite3
|
||||
import pymysql
|
||||
import pandas as pd
|
||||
from pathlib import Path
|
||||
from typing import Optional, List, Tuple
|
||||
from typing import Optional
|
||||
from contextlib import contextmanager
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
from sqlalchemy import create_engine, text
|
||||
|
||||
load_dotenv()
|
||||
|
||||
class DatabaseManager:
|
||||
"""Manages database for storing OHLCV data.
|
||||
Supports SQLite (local) and MySQL (remote).
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: str = "./data/market_data.db"):
|
||||
self.db_path = Path(db_path)
|
||||
self.use_mysql = bool(os.getenv('MYSQL_HOST'))
|
||||
|
||||
if not self.use_mysql:
|
||||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
class DatabaseManager:
|
||||
"""管理 OHLCV 数据的 MySQL 数据库"""
|
||||
|
||||
def __init__(self, db_path: str = None):
|
||||
"""初始化数据库管理器 (db_path 参数已废弃,仅为兼容性保留)"""
|
||||
self._engine = None
|
||||
self.initialize_db()
|
||||
|
||||
def _get_sqlalchemy_engine(self):
|
||||
"""获取 SQLAlchemy 引擎(用于 pandas read_sql)"""
|
||||
if self._engine is None:
|
||||
host = os.getenv('MYSQL_HOST')
|
||||
port = os.getenv('MYSQL_PORT', 3306)
|
||||
user = os.getenv('MYSQL_USER')
|
||||
password = os.getenv('MYSQL_PASSWORD')
|
||||
database = os.getenv('MYSQL_DB')
|
||||
self._engine = create_engine(
|
||||
f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset=utf8mb4"
|
||||
)
|
||||
return self._engine
|
||||
|
||||
@contextmanager
|
||||
def get_connection(self):
|
||||
"""Context manager for database connection."""
|
||||
if self.use_mysql:
|
||||
conn = pymysql.connect(
|
||||
host=os.getenv('MYSQL_HOST'),
|
||||
port=int(os.getenv('MYSQL_PORT', 3306)),
|
||||
user=os.getenv('MYSQL_USER'),
|
||||
password=os.getenv('MYSQL_PASSWORD'),
|
||||
database=os.getenv('MYSQL_DB'),
|
||||
charset='utf8mb4',
|
||||
# pandas requires standard cursor or handling
|
||||
)
|
||||
else:
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
|
||||
"""获取 MySQL 数据库连接"""
|
||||
conn = pymysql.connect(
|
||||
host=os.getenv('MYSQL_HOST'),
|
||||
port=int(os.getenv('MYSQL_PORT', 3306)),
|
||||
user=os.getenv('MYSQL_USER'),
|
||||
password=os.getenv('MYSQL_PASSWORD'),
|
||||
database=os.getenv('MYSQL_DB'),
|
||||
charset='utf8mb4',
|
||||
)
|
||||
try:
|
||||
yield conn
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def initialize_db(self):
|
||||
"""Create tables if they don't exist."""
|
||||
"""创建表(如果不存在)"""
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
if self.use_mysql:
|
||||
# MySQL Schema
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS ohlcv (
|
||||
timestamp DATETIME,
|
||||
symbol VARCHAR(20),
|
||||
timeframe VARCHAR(10),
|
||||
open DOUBLE,
|
||||
high DOUBLE,
|
||||
low DOUBLE,
|
||||
close DOUBLE,
|
||||
volume DOUBLE,
|
||||
PRIMARY KEY (symbol, timeframe, timestamp),
|
||||
INDEX idx_ohlcv_time (symbol, timeframe, timestamp)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
""")
|
||||
else:
|
||||
# SQLite Schema
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS ohlcv (
|
||||
timestamp DATETIME,
|
||||
symbol TEXT,
|
||||
timeframe TEXT,
|
||||
open REAL,
|
||||
high REAL,
|
||||
low REAL,
|
||||
close REAL,
|
||||
volume REAL,
|
||||
PRIMARY KEY (symbol, timeframe, timestamp)
|
||||
)
|
||||
""")
|
||||
cursor.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_ohlcv_time
|
||||
ON ohlcv (symbol, timeframe, timestamp)
|
||||
""")
|
||||
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS ohlcv (
|
||||
timestamp DATETIME,
|
||||
symbol VARCHAR(20),
|
||||
timeframe VARCHAR(10),
|
||||
open DOUBLE,
|
||||
high DOUBLE,
|
||||
low DOUBLE,
|
||||
close DOUBLE,
|
||||
volume DOUBLE,
|
||||
PRIMARY KEY (symbol, timeframe, timestamp),
|
||||
INDEX idx_ohlcv_time (symbol, timeframe, timestamp)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
def save_ohlcv(self, df: pd.DataFrame, symbol: str, timeframe: str):
|
||||
"""Save OHLCV data to database (Upsert)."""
|
||||
"""保存 OHLCV 数据到数据库 (Upsert)"""
|
||||
if df.empty:
|
||||
return
|
||||
|
||||
# Ensure timestamp is datetime
|
||||
# 确保 timestamp 是 datetime 类型
|
||||
df = df.copy()
|
||||
if not pd.api.types.is_datetime64_any_dtype(df['timestamp']):
|
||||
df['timestamp'] = pd.to_datetime(df['timestamp'])
|
||||
|
||||
# Convert to string for SQLite compatibility (ISO 8601) or MySQL
|
||||
# 转换为字符串格式
|
||||
df['timestamp'] = df['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
# Add metadata columns if they don't exist
|
||||
# 添加元数据列
|
||||
if 'symbol' not in df.columns:
|
||||
df['symbol'] = symbol
|
||||
if 'timeframe' not in df.columns:
|
||||
df['timeframe'] = timeframe
|
||||
|
||||
# Prepare data for insertion (ensure order)
|
||||
# records = df[['timestamp', 'symbol', 'timeframe', 'open', 'high', 'low', 'close', 'volume']].to_records(index=False)
|
||||
# to_records returns numpy record array, which sqlite3 adapts.
|
||||
# For MySQL/pymysql, we need list of tuples.
|
||||
|
||||
cols = ['timestamp', 'symbol', 'timeframe', 'open', 'high', 'low', 'close', 'volume']
|
||||
data = [tuple(row) for row in df[cols].values]
|
||||
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
if self.use_mysql:
|
||||
# MySQL uses %s and REPLACE INTO
|
||||
sql = """
|
||||
REPLACE INTO ohlcv
|
||||
(timestamp, symbol, timeframe, open, high, low, close, volume)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||||
"""
|
||||
cursor.executemany(sql, data)
|
||||
else:
|
||||
# SQLite uses ? and INSERT OR REPLACE
|
||||
sql = """
|
||||
INSERT OR REPLACE INTO ohlcv
|
||||
(timestamp, symbol, timeframe, open, high, low, close, volume)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
"""
|
||||
# sqlite3 accepts both list of tuples and numpy records, but list of tuples is safer for compatibility
|
||||
cursor.executemany(sql, data)
|
||||
|
||||
sql = """
|
||||
REPLACE INTO ohlcv
|
||||
(timestamp, symbol, timeframe, open, high, low, close, volume)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||||
"""
|
||||
cursor.executemany(sql, data)
|
||||
conn.commit()
|
||||
|
||||
def get_ohlcv(self, symbol: str, timeframe: str, start_date: Optional[str] = None, end_date: Optional[str] = None) -> pd.DataFrame:
|
||||
"""Retrieve OHLCV data from database."""
|
||||
"""从数据库获取 OHLCV 数据"""
|
||||
|
||||
placeholder = '%s' if self.use_mysql else '?'
|
||||
|
||||
query = f"""
|
||||
query = """
|
||||
SELECT timestamp, open, high, low, close, volume
|
||||
FROM ohlcv
|
||||
WHERE symbol = {placeholder} AND timeframe = {placeholder}
|
||||
WHERE symbol = :p0 AND timeframe = :p1
|
||||
"""
|
||||
params = [symbol, timeframe]
|
||||
params = {'p0': symbol, 'p1': timeframe}
|
||||
param_idx = 2
|
||||
|
||||
if start_date:
|
||||
query += f" AND timestamp >= {placeholder}"
|
||||
params.append(start_date)
|
||||
query += f" AND timestamp >= :p{param_idx}"
|
||||
params[f'p{param_idx}'] = start_date
|
||||
param_idx += 1
|
||||
|
||||
if end_date:
|
||||
# Handle 'YYYY-MM-DD' vs full timestamp
|
||||
if len(str(end_date)) == 10: # YYYY-MM-DD
|
||||
# SQLite has date(), MySQL uses DATE() or just string comparison
|
||||
if self.use_mysql:
|
||||
query += f" AND timestamp < DATE_ADD({placeholder}, INTERVAL 1 DAY)"
|
||||
else:
|
||||
query += f" AND timestamp < date({placeholder}, '+1 day')"
|
||||
# 处理 'YYYY-MM-DD' 格式
|
||||
if len(str(end_date)) == 10:
|
||||
query += f" AND timestamp < DATE_ADD(:p{param_idx}, INTERVAL 1 DAY)"
|
||||
else:
|
||||
query += f" AND timestamp <= {placeholder}"
|
||||
params.append(end_date)
|
||||
query += f" AND timestamp <= :p{param_idx}"
|
||||
params[f'p{param_idx}'] = end_date
|
||||
|
||||
query += " ORDER BY timestamp ASC"
|
||||
|
||||
with self.get_connection() as conn:
|
||||
df = pd.read_sql_query(query, conn, params=params, parse_dates=['timestamp'])
|
||||
engine = self._get_sqlalchemy_engine()
|
||||
with engine.connect() as conn:
|
||||
df = pd.read_sql_query(text(query), conn, params=params, parse_dates=['timestamp'])
|
||||
|
||||
return df
|
||||
|
||||
def get_latest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]:
|
||||
"""Get the latest timestamp for a symbol/timeframe."""
|
||||
placeholder = '%s' if self.use_mysql else '?'
|
||||
"""获取指定 symbol/timeframe 的最新时间戳"""
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(f"""
|
||||
cursor.execute("""
|
||||
SELECT MAX(timestamp)
|
||||
FROM ohlcv
|
||||
WHERE symbol = {placeholder} AND timeframe = {placeholder}
|
||||
WHERE symbol = %s AND timeframe = %s
|
||||
""", (symbol, timeframe))
|
||||
result = cursor.fetchone()
|
||||
|
||||
return pd.to_datetime(result[0]) if result and result[0] else None
|
||||
|
||||
def get_earliest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]:
|
||||
"""Get the earliest timestamp for a symbol/timeframe."""
|
||||
placeholder = '%s' if self.use_mysql else '?'
|
||||
"""获取指定 symbol/timeframe 的最早时间戳"""
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(f"""
|
||||
cursor.execute("""
|
||||
SELECT MIN(timestamp)
|
||||
FROM ohlcv
|
||||
WHERE symbol = {placeholder} AND timeframe = {placeholder}
|
||||
WHERE symbol = %s AND timeframe = %s
|
||||
""", (symbol, timeframe))
|
||||
result = cursor.fetchone()
|
||||
|
||||
|
||||
Binary file not shown.
+46
-115
@@ -1,4 +1,3 @@
|
||||
import sqlite3
|
||||
import pymysql
|
||||
from datetime import datetime
|
||||
import json
|
||||
@@ -8,43 +7,26 @@ from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
class TradingDatabase:
|
||||
"""交易数据库管理类 - 用于记录交易历史和预测结果
|
||||
"""交易数据库管理类 - 用于记录交易历史和预测结果 (MySQL)"""
|
||||
|
||||
支持 SQLite (本地) 和 MySQL (远程)。
|
||||
如果 .env 中配置了 MYSQL_HOST,则优先使用 MySQL。
|
||||
"""
|
||||
|
||||
def __init__(self, db_path="trading_history.db"):
|
||||
self.db_path = db_path
|
||||
self.use_mysql = bool(os.getenv('MYSQL_HOST'))
|
||||
def __init__(self, db_path=None):
|
||||
"""初始化 (db_path 参数已废弃,仅为兼容性保留)"""
|
||||
self._init_db()
|
||||
|
||||
@contextmanager
|
||||
def get_connection(self):
|
||||
"""数据库连接上下文管理器"""
|
||||
if self.use_mysql:
|
||||
conn = pymysql.connect(
|
||||
host=os.getenv('MYSQL_HOST'),
|
||||
port=int(os.getenv('MYSQL_PORT', 3306)),
|
||||
user=os.getenv('MYSQL_USER'),
|
||||
password=os.getenv('MYSQL_PASSWORD'),
|
||||
database=os.getenv('MYSQL_DB'),
|
||||
charset='utf8mb4',
|
||||
cursorclass=pymysql.cursors.DictCursor
|
||||
)
|
||||
else:
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
# Make sqlite behave like dict cursor for consistency if needed,
|
||||
# but existing code might expect tuple.
|
||||
# Actually existing code uses cursor.execute directly.
|
||||
# SQLite cursor returns tuples by default.
|
||||
# PyMySQL DictCursor returns dicts.
|
||||
# We need to be careful about compatibility.
|
||||
# The existing log_trade uses INSERT which is fine.
|
||||
# check_existing_trade uses SELECT count(*), fetchone()[0], which works for both if we don't use DictCursor for MySQL or handle it.
|
||||
# Let's use standard cursor for MySQL for compatibility with existing fetchone()[0] usage.
|
||||
|
||||
"""获取 MySQL 数据库连接"""
|
||||
conn = pymysql.connect(
|
||||
host=os.getenv('MYSQL_HOST'),
|
||||
port=int(os.getenv('MYSQL_PORT', 3306)),
|
||||
user=os.getenv('MYSQL_USER'),
|
||||
password=os.getenv('MYSQL_PASSWORD'),
|
||||
database=os.getenv('MYSQL_DB'),
|
||||
charset='utf8mb4',
|
||||
cursorclass=pymysql.cursors.DictCursor
|
||||
)
|
||||
try:
|
||||
yield conn
|
||||
finally:
|
||||
@@ -54,81 +36,45 @@ class TradingDatabase:
|
||||
"""初始化数据库表结构"""
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
if self.use_mysql:
|
||||
# MySQL Schema
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS trades (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
timestamp DATETIME,
|
||||
market_title VARCHAR(255),
|
||||
market_slug VARCHAR(255),
|
||||
model_name VARCHAR(50),
|
||||
prediction VARCHAR(10),
|
||||
probability DOUBLE,
|
||||
action VARCHAR(10),
|
||||
price DOUBLE,
|
||||
amount DOUBLE,
|
||||
size DOUBLE,
|
||||
status VARCHAR(20),
|
||||
error_message TEXT,
|
||||
raw_data JSON,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
INDEX idx_market_slug (market_slug),
|
||||
INDEX idx_status (status)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
''')
|
||||
else:
|
||||
# SQLite Schema
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS trades (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
timestamp TEXT,
|
||||
market_title TEXT,
|
||||
market_slug TEXT,
|
||||
model_name TEXT,
|
||||
prediction TEXT,
|
||||
probability REAL,
|
||||
action TEXT,
|
||||
price REAL,
|
||||
amount REAL,
|
||||
size REAL,
|
||||
status TEXT,
|
||||
error_message TEXT,
|
||||
raw_data TEXT,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
''')
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS trades (
|
||||
id INT AUTO_INCREMENT PRIMARY KEY,
|
||||
timestamp DATETIME,
|
||||
market_title VARCHAR(255),
|
||||
market_slug VARCHAR(255),
|
||||
model_name VARCHAR(50),
|
||||
prediction VARCHAR(10),
|
||||
probability DOUBLE,
|
||||
action VARCHAR(10),
|
||||
price DOUBLE,
|
||||
amount DOUBLE,
|
||||
size DOUBLE,
|
||||
status VARCHAR(20),
|
||||
error_message TEXT,
|
||||
raw_data JSON,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
INDEX idx_market_slug (market_slug),
|
||||
INDEX idx_status (status)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
''')
|
||||
conn.commit()
|
||||
|
||||
def log_trade(self, trade_data: dict):
|
||||
"""
|
||||
记录交易信息
|
||||
"""
|
||||
"""记录交易信息"""
|
||||
try:
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Prepare data
|
||||
timestamp = trade_data.get('timestamp', datetime.now().isoformat())
|
||||
raw_data_json = json.dumps(trade_data.get('raw_data', {}), ensure_ascii=False)
|
||||
|
||||
if self.use_mysql:
|
||||
sql = '''
|
||||
INSERT INTO trades (
|
||||
timestamp, market_title, market_slug, model_name,
|
||||
prediction, probability, action, price, amount, size,
|
||||
status, error_message, raw_data
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
'''
|
||||
else:
|
||||
sql = '''
|
||||
INSERT INTO trades (
|
||||
timestamp, market_title, market_slug, model_name,
|
||||
prediction, probability, action, price, amount, size,
|
||||
status, error_message, raw_data
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
'''
|
||||
sql = '''
|
||||
INSERT INTO trades (
|
||||
timestamp, market_title, market_slug, model_name,
|
||||
prediction, probability, action, price, amount, size,
|
||||
status, error_message, raw_data
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
'''
|
||||
|
||||
cursor.execute(sql, (
|
||||
timestamp,
|
||||
@@ -150,29 +96,14 @@ class TradingDatabase:
|
||||
print(f"写入数据库失败: {e}")
|
||||
|
||||
def check_existing_trade(self, market_slug: str, status: str = 'success') -> bool:
|
||||
"""
|
||||
检查指定市场是否已有成功交易记录
|
||||
"""
|
||||
"""检查指定市场是否已有成功交易记录"""
|
||||
try:
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
if self.use_mysql:
|
||||
sql = 'SELECT COUNT(*) FROM trades WHERE market_slug = %s AND status = %s'
|
||||
else:
|
||||
sql = 'SELECT COUNT(*) FROM trades WHERE market_slug = ? AND status = ?'
|
||||
|
||||
sql = 'SELECT COUNT(*) as cnt FROM trades WHERE market_slug = %s AND status = %s'
|
||||
cursor.execute(sql, (market_slug, status))
|
||||
|
||||
# Fetch result compatible with both
|
||||
result = cursor.fetchone()
|
||||
if isinstance(result, dict): # If DictCursor was used (not used here but for safety)
|
||||
count = list(result.values())[0]
|
||||
elif isinstance(result, (list, tuple)):
|
||||
count = result[0]
|
||||
else:
|
||||
count = 0
|
||||
|
||||
count = result['cnt'] if result else 0
|
||||
return count > 0
|
||||
except Exception as e:
|
||||
print(f"检查已存在交易失败: {e}")
|
||||
|
||||
Reference in new issue
Block a user