feat: upgrade to multi-model ensemble system

- Updated DB schemas: added 'strategy' to trades, new 'hourly_model_predictions' wide table
- Refactored backend: api/models.py, api/prediction_service.py to support wide format
- Updated trader: auto_trade_polymarket.py now logs strategy and saves to new table
- Updated frontend: index.html and app.js to display multi-model accuracy and trade strategy
This commit is contained in:
crypto_predictor committed 2026-02-11 22:13:35 +08:00
1 parent c0a02868ac
commit 2993014fa8
5 files changed
+465 -335

No files matched your search

+277 -192
View File
@@ -88,6 +88,7 @@ def _dict_from_row(row):
def init_model_predictions_table():
"""初始化 model_predictions 表"""
# 兼容旧表,但主要使用 hourly_model_predictions
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
@@ -110,6 +111,42 @@ def init_model_predictions_table():
INDEX idx_model (model_name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
''')
# 确保 hourly_model_predictions 表存在
cursor.execute('''
CREATE TABLE IF NOT EXISTS hourly_model_predictions (
id INT AUTO_INCREMENT PRIMARY KEY,
timestamp DATETIME NOT NULL,
timestamp_et VARCHAR(50),
date VARCHAR(10),
hour INT,
open_price DOUBLE,
close_price DOUBLE,
actual_result INT COMMENT '0=Down, 1=Up',
dlinear_prediction INT,
dlinear_prob DOUBLE,
dlinear_correct INT,
nbeats_prediction INT,
nbeats_prob DOUBLE,
nbeats_correct INT,
ttm_prediction INT,
ttm_prob DOUBLE,
ttm_correct INT,
ensemble_prediction INT,
ensemble_votes VARCHAR(50),
ensemble_correct INT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uk_date_hour (date, hour),
INDEX idx_timestamp (timestamp)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
''')
conn.commit()
@@ -143,12 +180,6 @@ class TradeModel:
day: Optional[str] = None) -> Dict:
"""
获取历史交易记录
Args:
page: 页码
limit: 每页条数
month: 月份筛选 (YYYY-MM)
day: 日期筛选 (YYYY-MM-DD)
"""
offset = (page - 1) * limit
@@ -160,7 +191,7 @@ class TradeModel:
params = []
if month:
where_clauses.append("DATE_FORMAT(timestamp, '%%Y-%%m') = %s")
where_clauses.append("DATE_FORMAT(timestamp, '%Y-%m') = %s")
params.append(month)
if day:
@@ -199,274 +230,328 @@ class TradeModel:
class PredictionModel:
"""预测记录模型"""
"""预测记录模型 (V2 - 宽表结构)"""
@staticmethod
def get_prediction(date: str, hour: int, model_name: str) -> Optional[Dict]:
"""获取指定时间和模型的预测记录"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM model_predictions
WHERE date = %s AND hour = %s AND model_name = %s
""", (date, hour, model_name))
row = cursor.fetchone()
return _dict_from_row(row)
@staticmethod
def save_prediction(data: Dict) -> int:
def save_hourly_prediction(data: Dict) -> int:
"""
保存预测记录
Args:
data: 预测数据字典
Returns:
插入的记录ID
保存每小时综合预测记录 (宽表)
"""
with get_db_connection() as conn:
cursor = conn.cursor()
sql = """
INSERT INTO model_predictions
(timestamp, timestamp_et, date, hour, model_name,
prediction, probability, actual_result, is_correct,
open_price, close_price)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
INSERT INTO hourly_model_predictions
(timestamp, timestamp_et, date, hour, open_price, close_price, actual_result,
dlinear_prediction, dlinear_prob, dlinear_correct,
nbeats_prediction, nbeats_prob, nbeats_correct,
ttm_prediction, ttm_prob, ttm_correct,
ensemble_prediction, ensemble_votes, ensemble_correct)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
prediction = VALUES(prediction),
probability = VALUES(probability),
actual_result = VALUES(actual_result),
is_correct = VALUES(is_correct),
open_price = VALUES(open_price),
close_price = VALUES(close_price)
close_price = VALUES(close_price),
actual_result = VALUES(actual_result),
dlinear_prediction = VALUES(dlinear_prediction),
dlinear_prob = VALUES(dlinear_prob),
dlinear_correct = VALUES(dlinear_correct),
nbeats_prediction = VALUES(nbeats_prediction),
nbeats_prob = VALUES(nbeats_prob),
nbeats_correct = VALUES(nbeats_correct),
ttm_prediction = VALUES(ttm_prediction),
ttm_prob = VALUES(ttm_prob),
ttm_correct = VALUES(ttm_correct),
ensemble_prediction = VALUES(ensemble_prediction),
ensemble_votes = VALUES(ensemble_votes),
ensemble_correct = VALUES(ensemble_correct)
"""
cursor.execute(sql, (
data.get('timestamp'),
data.get('timestamp_et'),
data.get('date'),
data.get('hour'),
data.get('model_name'),
data.get('prediction'),
data.get('probability'),
data.get('actual_result'),
data.get('is_correct'),
data.get('open_price'),
data.get('close_price')
data.get('close_price'),
data.get('actual_result'),
data.get('dlinear_prediction'),
data.get('dlinear_prob'),
data.get('dlinear_correct'),
data.get('nbeats_prediction'),
data.get('nbeats_prob'),
data.get('nbeats_correct'),
data.get('ttm_prediction'),
data.get('ttm_prob'),
data.get('ttm_correct'),
data.get('ensemble_prediction'),
str(data.get('ensemble_votes')) if data.get('ensemble_votes') else None,
data.get('ensemble_correct')
))
conn.commit()
return cursor.lastrowid or 0
@staticmethod
def update_actual_result(date: str, hour: int, model_name: str,
actual_result: int, close_price: float) -> bool:
"""更新实际结果"""
def update_actual_result_v2(date: str, hour: int, actual_result: int, close_price: float) -> bool:
"""
更新实际结果,并重新计算各模型正确性 (V2)
"""
with get_db_connection() as conn:
cursor = conn.cursor()
# 先获取预测值
# 获取该小时记录
cursor.execute("""
SELECT prediction FROM model_predictions
WHERE date = %s AND hour = %s AND model_name = %s
""", (date, hour, model_name))
SELECT * FROM hourly_model_predictions
WHERE date = %s AND hour = %s
""", (date, hour))
row = cursor.fetchone()
if not row:
return False
prediction = row['prediction']
is_correct = 1 if prediction == actual_result else 0
# 计算正确性
updates = {
'actual_result': actual_result,
'close_price': close_price
}
cursor.execute("""
UPDATE model_predictions
SET actual_result = %s, is_correct = %s, close_price = %s
WHERE date = %s AND hour = %s AND model_name = %s
""", (actual_result, is_correct, close_price, date, hour, model_name))
# DLinear
if row['dlinear_prediction'] is not None:
updates['dlinear_correct'] = 1 if row['dlinear_prediction'] == actual_result else 0
# N-BEATS
if row['nbeats_prediction'] is not None:
updates['nbeats_correct'] = 1 if row['nbeats_prediction'] == actual_result else 0
# TTM
if row['ttm_prediction'] is not None:
updates['ttm_correct'] = 1 if row['ttm_prediction'] == actual_result else 0
# Ensemble
if row['ensemble_prediction'] is not None:
updates['ensemble_correct'] = 1 if row['ensemble_prediction'] == actual_result else 0
# 构建更新SQL
set_clause = ", ".join([f"{k} = %s" for k in updates.keys()])
sql = f"UPDATE hourly_model_predictions SET {set_clause} WHERE date = %s AND hour = %s"
cursor.execute(sql, list(updates.values()) + [date, hour])
conn.commit()
return True
@staticmethod
def get_monthly_stats(month: str) -> Dict:
def get_monthly_stats_v2(month: str) -> Dict:
"""
获取月度统计
Args:
month: 月份 (YYYY-MM)
Returns:
各模型的统计数据
获取月度统计 (V2 - 从宽表)
"""
with get_db_connection() as conn:
cursor = conn.cursor()
sql = """
SELECT
model_name,
COUNT(*) as total,
SUM(CASE WHEN is_correct = 1 THEN 1 ELSE 0 END) as correct,
SUM(CASE WHEN is_correct = 0 THEN 1 ELSE 0 END) as incorrect
FROM model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
AND actual_result IS NOT NULL
GROUP BY model_name
"""
cursor.execute(sql, (month,))
rows = cursor.fetchall()
# 统计各个模型的准确率
models = ['dlinear', 'nbeats', 'ttm', 'ensemble']
result = {}
for row in rows:
row_dict = _dict_from_row(row)
model = row_dict['model_name']
total = row_dict['total']
correct = row_dict['correct'] or 0
incorrect = row_dict['incorrect'] or 0
accuracy = (correct / total * 100) if total > 0 else 0
for model in models:
col_pred = f"{model}_prediction"
col_correct = f"{model}_correct"
result[model] = {
"total": total,
"correct": correct,
"incorrect": incorrect,
"accuracy": round(accuracy, 2)
}
sql = f"""
SELECT
COUNT({col_pred}) as total,
SUM(CASE WHEN {col_correct} = 1 THEN 1 ELSE 0 END) as correct,
SUM(CASE WHEN {col_correct} = 0 THEN 1 ELSE 0 END) as incorrect
FROM hourly_model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
AND actual_result IS NOT NULL
"""
cursor.execute(sql, (month,))
row = cursor.fetchone()
if row:
total = row['total']
correct = row['correct'] or 0
incorrect = row['incorrect'] or 0
accuracy = (correct / total * 100) if total > 0 else 0
# 格式化键名以匹配前端 (例如 dlinear -> DLinear)
model_key = {
'dlinear': 'DLinear',
'nbeats': 'N-BEATS',
'ttm': 'TTM-R2',
'ensemble': 'Ensemble'
}.get(model, model)
result[model_key] = {
"total": total,
"correct": correct,
"incorrect": incorrect,
"accuracy": round(accuracy, 2)
}
return result
@staticmethod
def get_daily_stats(month: str, page: int = 1, limit: int = 10) -> Dict:
def get_daily_stats_v2(month: str, page: int = 1, limit: int = 10) -> Dict:
"""
获取每日统计
Args:
month: 月份 (YYYY-MM)
page: 页码
limit: 每页条数
获取每日统计 (V2 - 从宽表)
"""
with get_db_connection() as conn:
cursor = conn.cursor()
sql = """
SELECT
DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as date_cst,
model_name,
COUNT(*) as total,
SUM(CASE WHEN is_correct = 1 THEN 1 ELSE 0 END) as correct,
SUM(CASE WHEN is_correct = 0 THEN 1 ELSE 0 END) as incorrect
FROM model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
AND actual_result IS NOT NULL
GROUP BY date_cst, model_name
ORDER BY date_cst DESC
"""
cursor.execute(sql, (month,))
rows = cursor.fetchall()
# 重组数据: 按日期合并不同模型的统计
daily_data = {}
for row in rows:
row_dict = _dict_from_row(row)
date = str(row_dict['date_cst']) # CST 日期
if date not in daily_data:
daily_data[date] = {"date": date}
model = row_dict['model_name']
total = row_dict['total']
correct = row_dict['correct'] or 0
incorrect = row_dict['incorrect'] or 0
accuracy = (correct / total * 100) if total > 0 else 0
daily_data[date][model] = {
"total": total,
"correct": correct,
"incorrect": incorrect,
"accuracy": round(accuracy, 2)
}
# 排序并分页
sorted_dates = sorted(daily_data.keys(), reverse=True)
total_count = len(sorted_dates)
# 分页查询日期
offset = (page - 1) * limit
paginated_dates = sorted_dates[offset:offset + limit]
date_sql = """
SELECT DISTINCT DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as date_cst
FROM hourly_model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
ORDER BY date_cst DESC
LIMIT %s OFFSET %s
"""
cursor.execute(date_sql, (month, limit, offset))
dates = [row['date_cst'] for row in cursor.fetchall()]
data = [daily_data[d] for d in paginated_dates]
# 总天数
count_sql = """
SELECT COUNT(DISTINCT DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00'))) as cnt
FROM hourly_model_predictions
WHERE DATE_FORMAT(CONVERT_TZ(timestamp, '+00:00', '+08:00'), '%%Y-%%m') = %s
"""
cursor.execute(count_sql, (month,))
total_count = cursor.fetchone()['cnt']
daily_data = []
models = ['dlinear', 'nbeats', 'ttm', 'ensemble']
for date_obj in dates:
date_str = str(date_obj)
day_stats = {"date": date_str}
for model in models:
col_pred = f"{model}_prediction"
col_correct = f"{model}_correct"
sql = f"""
SELECT
COUNT({col_pred}) as total,
SUM(CASE WHEN {col_correct} = 1 THEN 1 ELSE 0 END) as correct
FROM hourly_model_predictions
WHERE DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) = %s
AND actual_result IS NOT NULL
"""
cursor.execute(sql, (date_str,))
row = cursor.fetchone()
total = row['total']
correct = row['correct'] or 0
accuracy = (correct / total * 100) if total > 0 else 0
model_key = {
'dlinear': 'DLinear',
'nbeats': 'N-BEATS',
'ttm': 'TTM-R2',
'ensemble': 'Ensemble'
}.get(model, model)
day_stats[model_key] = {
"total": total,
"correct": correct,
"accuracy": round(accuracy, 2)
}
daily_data.append(day_stats)
return {
"data": data,
"data": daily_data,
"page": page,
"limit": limit,
"total": total_count,
"total_pages": (total_count + limit - 1) // limit if limit > 0 else 0
}
@staticmethod
def get_hourly_details(date: str) -> List[Dict]:
def get_hourly_details_v2(date: str) -> List[Dict]:
"""
获取指定日期的每小时详情
Args:
date: 日期 (YYYY-MM-DD, CST 北京时间)
获取指定日期的每小时详情 (V2)
"""
with get_db_connection() as conn:
cursor = conn.cursor()
# 按 CST 日期筛选,并获取 CST 小时
cursor.execute("""
SELECT
*,
HOUR(CONVERT_TZ(timestamp, '+00:00', '+08:00')) as hour_cst,
CONVERT_TZ(timestamp, '+00:00', '+08:00') as timestamp_cst_dt
FROM model_predictions
FROM hourly_model_predictions
WHERE DATE(CONVERT_TZ(timestamp, '+00:00', '+08:00')) = %s
ORDER BY hour_cst, model_name
ORDER BY hour_cst
""", (date,))
rows = cursor.fetchall()
result = []
# 按 CST 小时重组数据
hourly_data = {}
for row in rows:
row_dict = _dict_from_row(row)
hour_cst = row_dict['hour_cst'] # 使用 CST 小时
if hour_cst not in hourly_data:
# 将 timestamp 转换为 CST 时间字符串
timestamp_cst = convert_to_cst(row_dict.get('timestamp', ''))
hourly_data[hour_cst] = {
"hour": hour_cst, # 返回 CST 小时
"timestamp_et": row_dict['timestamp_et'],
"timestamp_cst": timestamp_cst,
"open_price": row_dict['open_price'],
"close_price": row_dict['close_price'],
"actual_result": row_dict['actual_result'],
"models": {}
hour_data = {
"hour": row['hour_cst'],
"timestamp_et": row['timestamp_et'],
"timestamp_cst": convert_to_cst(row.get('timestamp', '')),
"open_price": row['open_price'],
"close_price": row['close_price'],
"actual_result": row['actual_result'],
"models": {
"DLinear": {
"prediction": row['dlinear_prediction'],
"probability": row['dlinear_prob'],
"is_correct": row['dlinear_correct']
},
"N-BEATS": {
"prediction": row['nbeats_prediction'],
"probability": row['nbeats_prob'],
"is_correct": row['nbeats_correct']
},
"TTM-R2": {
"prediction": row['ttm_prediction'],
"probability": row['ttm_prob'],
"is_correct": row['ttm_correct']
},
"Ensemble": {
"prediction": row['ensemble_prediction'],
"is_correct": row['ensemble_correct'],
"votes": row['ensemble_votes']
}
}
model = row_dict['model_name']
hourly_data[hour_cst]["models"][model] = {
"prediction": row_dict['prediction'],
"probability": row_dict['probability'],
"is_correct": row_dict['is_correct']
}
# 按 CST 小时排序返回
return [hourly_data[h] for h in sorted(hourly_data.keys())]
result.append(hour_data)
return result
@staticmethod
def save_prediction(data: Dict) -> int:
"""兼容旧接口 - 如果需要的话,暂时保留,或者可以将旧表写入也放在这里"""
# 旧表写入逻辑... 略,或保持不变
# 为了完整性,我们可以保留对旧表的操作,或者逐步弃用。
# 这里为了安全,保留原方法的签名,实际调用旧表逻辑。
# 但我们主要逻辑将切换到 save_hourly_prediction
# 见 prediction_service.py 的修改
# ... (keep old logic or redirect) ...
# 这里我们假设 prediction_service.py 会改为调用 save_hourly_prediction
# 如果还有地方调用 save_prediction (比如测试代码), 我们可以保留它
return 0
@staticmethod
def get_missing_hours(date: str, model_name: str) -> List[int]:
def update_actual_result(date: str, hour: int, model_name: str,
actual_result: int, close_price: float) -> bool:
"""兼容旧接口"""
return False
@staticmethod
def get_missing_hours(date: str, model_name: str = 'Ensemble') -> List[int]:
"""
获取指定日期缺少预测的小时列表
Args:
date: 日期 (YYYY-MM-DD)
model_name: 模型名称
Returns:
缺少预测的小时列表
获取指定日期缺少预测的小时列表 (V2)
model_name 参数现在主要用于兼容,实际检查整行是否存在
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT hour FROM model_predictions
WHERE date = %s AND model_name = %s
""", (date, model_name))
SELECT hour FROM hourly_model_predictions
WHERE date = %s
""", (date,))
rows = cursor.fetchall()
existing_hours = {row['hour'] for row in rows}