215 lines
7.3 KiB
Python
215 lines
7.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
||
from sqlalchemy.orm import Session
|
||
from typing import Dict
|
||
from ...core.database import get_db
|
||
from ...services.prediction_service import PredictionService
|
||
from ...models.lottery import SSQLotteryBetRecord, DLTLotteryBetRecord, SSQLottery, DLTLottery
|
||
import datetime
|
||
import random
|
||
from ...services.telegram_bot import send_message
|
||
from ...services.analysis_service import LotteryAnalysisService
|
||
import os
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.post("/train/{lottery_type}")
|
||
def train_prediction_model(
|
||
lottery_type: str,
|
||
periods: int = 100,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""训练预测模型"""
|
||
if lottery_type not in ['ssq', 'dlt']:
|
||
raise HTTPException(status_code=400, detail="Invalid lottery type")
|
||
|
||
service = PredictionService(db)
|
||
return service.train_model(lottery_type, periods)
|
||
|
||
|
||
@router.get("/predict/{lottery_type}")
|
||
def predict_next_numbers(
|
||
lottery_type: str,
|
||
periods: int = 10,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""预测下一期号码"""
|
||
if lottery_type not in ['ssq', 'dlt']:
|
||
raise HTTPException(status_code=400, detail="Invalid lottery type")
|
||
|
||
service = PredictionService(db)
|
||
return service.predict_next_numbers(lottery_type, periods)
|
||
|
||
|
||
@router.get("/pattern/{lottery_type}")
|
||
def get_pattern_prediction(
|
||
lottery_type: str,
|
||
periods: int = 100,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""基于模式的预测"""
|
||
if lottery_type not in ['ssq', 'dlt']:
|
||
raise HTTPException(status_code=400, detail="Invalid lottery type")
|
||
|
||
service = PredictionService(db)
|
||
return service.get_pattern_based_prediction(lottery_type, periods)
|
||
|
||
|
||
@router.get("/ensemble/{lottery_type}")
|
||
def get_ensemble_prediction(
|
||
lottery_type: str,
|
||
periods: int = 100,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
"""集成预测"""
|
||
if lottery_type not in ['ssq', 'dlt']:
|
||
raise HTTPException(status_code=400, detail="Invalid lottery type")
|
||
|
||
service = PredictionService(db)
|
||
return service.get_ensemble_prediction(lottery_type, periods)
|
||
|
||
|
||
@router.post("/recommend/today")
|
||
def recommend_today(db: Session = Depends(get_db)):
|
||
"""
|
||
每日推荐：一三六推荐大乐透，二四日推荐双色球，固定投注1注+推荐4注，共5注，推送内容合并快乐8。
|
||
"""
|
||
# 固定投注号码（写死）
|
||
FIXED_SSQ = {'red': [4, 6, 7, 12, 18, 19], 'blue': [9]}
|
||
FIXED_DLT = {'red': [4, 6, 9, 18, 19], 'blue': [7, 12]}
|
||
# 判断彩种
|
||
weekday = datetime.datetime.now().isoweekday()
|
||
if weekday in [1, 3, 6]:
|
||
lottery_type = 'dlt'
|
||
BetModel = DLTLotteryBetRecord
|
||
recommend_type = '集成预测-每日推荐'
|
||
|
||
def number_format(red, blue): return ' '.join(
|
||
f"{n:02d}" for n in red) + ' 蓝球：' + ' '.join(f"{n:02d}" for n in blue)
|
||
LotteryModel = DLTLottery
|
||
fixed = FIXED_DLT
|
||
title = "你好，帮忙买下大乐透。"
|
||
dlt_append = True
|
||
else:
|
||
lottery_type = 'ssq'
|
||
BetModel = SSQLotteryBetRecord
|
||
recommend_type = '集成预测-每日推荐'
|
||
|
||
def number_format(red, blue): return ' '.join(
|
||
f"{n:02d}" for n in red) + ' 蓝球：' + ' '.join(f"{n:02d}" for n in blue)
|
||
LotteryModel = SSQLottery
|
||
fixed = FIXED_SSQ
|
||
title = "你好，帮忙买下双色球。"
|
||
dlt_append = False
|
||
# 生成batch_id
|
||
today = datetime.datetime.now().strftime('%Y%m%d')
|
||
batch_id = f"{today}-{random.randint(1000, 9999)}"
|
||
# 1. 推荐4注（集成预测+智能选号补足）
|
||
service = PredictionService(db)
|
||
pred = service.get_ensemble_prediction(lottery_type, periods=200)
|
||
if not pred.get('success'):
|
||
raise HTTPException(
|
||
status_code=500, detail=pred.get('message', '预测失败'))
|
||
recs = pred.get('recommendations', [])
|
||
results = []
|
||
used_numbers = set()
|
||
# 先取集成预测推荐
|
||
for rec in recs:
|
||
if len(results) >= 4:
|
||
break
|
||
red = rec.get('numbers')
|
||
blue = [rec.get('blue')] if rec.get(
|
||
'blue') is not None else rec.get('blues', [])
|
||
if not red or not blue:
|
||
continue
|
||
numbers = ','.join(f"{n:02d}" for n in red) + \
|
||
'|' + ','.join(f"{n:02d}" for n in blue)
|
||
if numbers in used_numbers:
|
||
continue
|
||
used_numbers.add(numbers)
|
||
last = db.query(LotteryModel).order_by(
|
||
LotteryModel.open_time.desc()).first()
|
||
issue = last.issue if last else ''
|
||
bet = BetModel(
|
||
batch_id=batch_id,
|
||
issue=issue,
|
||
numbers=numbers,
|
||
recommend_type=rec.get('method', recommend_type)
|
||
)
|
||
db.add(bet)
|
||
db.commit()
|
||
db.refresh(bet)
|
||
results.append({
|
||
'id': bet.id,
|
||
'batch_id': batch_id,
|
||
'issue': issue,
|
||
'numbers': numbers,
|
||
'red': red,
|
||
'blue': blue,
|
||
'recommend_type': rec.get('method', recommend_type)
|
||
})
|
||
# 补足到4注
|
||
analysis_service = LotteryAnalysisService(db)
|
||
while len(results) < 4:
|
||
smart = analysis_service.generate_smart_numbers(
|
||
lottery_type, strategy='balanced', count=1)
|
||
if not smart:
|
||
break
|
||
red = smart[0].get('red_numbers')
|
||
blue = smart[0].get('blue_numbers')
|
||
numbers = ','.join(f"{n:02d}" for n in red) + \
|
||
'|' + ','.join(f"{n:02d}" for n in blue)
|
||
if numbers in used_numbers:
|
||
continue
|
||
used_numbers.add(numbers)
|
||
last = db.query(LotteryModel).order_by(
|
||
LotteryModel.open_time.desc()).first()
|
||
issue = last.issue if last else ''
|
||
bet = BetModel(
|
||
batch_id=batch_id,
|
||
issue=issue,
|
||
numbers=numbers,
|
||
recommend_type='智能选号补足'
|
||
)
|
||
db.add(bet)
|
||
db.commit()
|
||
db.refresh(bet)
|
||
results.append({
|
||
'id': bet.id,
|
||
'batch_id': batch_id,
|
||
'issue': issue,
|
||
'numbers': numbers,
|
||
'red': red,
|
||
'blue': blue,
|
||
'recommend_type': '智能选号补足'
|
||
})
|
||
# 2. 组装推送内容（固定投注+4注推荐）
|
||
msg_lines = [title]
|
||
# 固定投注
|
||
msg_lines.append(f"红球：{' '.join(f'{n:02d}' for n in fixed['red'])} 蓝球：{
|
||
' '.join(f'{n:02d}' for n in fixed['blue'])}")
|
||
# 推荐号码
|
||
for rec in results:
|
||
msg_lines.append(f"红球：{' '.join(f'{n:02d}' for n in rec['red'])} 蓝球：{
|
||
' '.join(f'{n:02d}' for n in rec['blue'])}")
|
||
if dlt_append:
|
||
msg_lines.append("都追加，谢谢。")
|
||
# 3. 快乐8推送（3注十选，合并到主消息）
|
||
msg_lines.append("\n快乐8，单式，选十")
|
||
for _ in range(3):
|
||
nums = sorted(random.sample(range(1, 81), 10))
|
||
msg_lines.append(' '.join(f"{n:02d}" for n in nums))
|
||
# 4. 推送到Telegram
|
||
try:
|
||
token = os.getenv("TELEGRAM_BOT_TOKEN")
|
||
chat_id = os.getenv("TELEGRAM_CHAT_ID")
|
||
send_message('\n'.join(msg_lines), token=token, chat_id=chat_id)
|
||
except Exception as e:
|
||
print(f"[WARN] Telegram推送失败: {e}")
|
||
return {
|
||
'success': True,
|
||
'lottery_type': lottery_type,
|
||
'batch_id': batch_id,
|
||
'recommend': results
|
||
}
|