liwq/bocai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

调整py脚本

Browse Source
This commit is contained in:
liwq
2026-01-28 09:44:21 +08:00
parent 639ea86e37
commit edebf424db
191 changed files with 3767620 additions and 60489 deletions
Download Patch File Download Diff File Expand all files Collapse all files

278
PyModel/predict_model_v14_profit_maximizer.py Normal file
View File

@@ -0,0 +1,278 @@
import pandas as pd
import numpy as np
import json
import configparser
from datetime import datetime, timedelta
conf = configparser.ConfigParser()
# 读取配置文件
conf.read('conf.ini', encoding='utf-8')
class LotteryPredictorV14Profit:
"""
V28 "Iron-Shield" (铁盾终极版)
核心逻辑: V27 动态流水增强 (3-8注) + 绝佳 30% (6000元) 日内止损熔断。
目标: 100元/注重仓下，最高收益/回撤比。
"""
# --- 核心参数 ---
ODDS = float(conf["init"]["ODDS"])
REBATE_RATE = float(conf["dynamic"]["REBATE_RATE"])
BASE_BET_UNIT = int(conf["init"]["BASE_BET_UNIT"])
_bet_this_round = int(conf["init"]["BET_THIS_ROUND"])
SAFE_BET_LEVEL_1 = 0.05
SAFE_BET_LEVEL_2 = 0.10
SAFE_BET_LEVEL_3 = 0.15
SAFE_BET_LEVEL_4 = 0.20
SAFE_BET_LEVEL_5 = 0.25
RADICAL_BET_LEVEL_0 = 0.05
RADICAL_BET_LEVEL_1 = 0.1
RADICAL_BET_LEVEL_2 = 0.2
INITIAL_CAPITAL = int(conf["init"]["INITIAL_CAPITAL"])
DAILY_STOP_LOSS_RATE = float(conf["init"]["DAILY_STOP_LOSS_RATE"]) # 绝佳熔断点：6000元熔断
POSITION_LOOKBACK_PERIODS = int(conf["dynamic"]["POSITION_LOOKBACK_PERIODS"])
OMISSIONS_LEVEL_1 = int(conf["dynamic"]["OMISSIONS_LEVEL_1"])
OMISSIONS_LEVEL_2 = int(conf["dynamic"]["OMISSIONS_LEVEL_2"])
OMISSIONS_LEVEL_3 = int(conf["dynamic"]["OMISSIONS_LEVEL_3"])
MOMENTUM_LEVEL_1 = int(conf["dynamic"]["MOMENTUM_LEVEL_1"])
MOMENTUM_LEVEL_2 = int(conf["dynamic"]["MOMENTUM_LEVEL_2"])
MOMENTUM_LEVEL_3 = int(conf["dynamic"]["MOMENTUM_LEVEL_3"])
RECENT_OMISSIONS = int(conf["dynamic"]["RECENT_OMISSIONS"])
RECENT_MOMENTUM = int(conf["dynamic"]["RECENT_MOMENTUM"])
# V27 分级投注参数
# S级: 遗漏>=55, 动能>=45 -> 8注
# A级: 遗漏>=50, 动能>=40 -> 6注
# B级: 遗漏>=45, 动能>=35 -> 4注
def __init__(self):
# 用于追踪当日盈亏，确保日内止损
# {'pnl': 0, 'bets': 0, 'miss_count': 0}
self.daily_pnl_tracker = {}
def _calculate_omission_and_momentum(self, history_data):
"""计算遗漏和动能"""
results = [record for record in history_data["result"]]
results_array = np.array(results)
if len(results_array) < 500:
return None, None
recent_400 = results_array[-400:]
recent_500 = results_array[-500:]
omissions = np.full((10, 10), 400)
# 计算遗漏
for pos in range(10):
for num in range(1, 11):
found = np.where(recent_400[:, pos] == num)[0]
if len(found) > 0:
omissions[pos, num - 1] = len(recent_400) - 1 - found[-1]
else:
omissions[pos, num - 1] = 400
# 计算动能 (最近 500 期出现次数)
momentum = np.zeros((10, 10))
for pos in range(10):
for num in range(1, 11):
momentum[pos, num - 1] = np.sum(recent_500[:, pos] == num)
return omissions, momentum
def _get_dynamic_bet_nums(self, pos, omissions, momentum):
"""根据信号强度动态决定投注注数和号码"""
scores = []
for n in range(1, 11):
o = omissions[pos, n - 1]
m = momentum[pos, n - 1]
if o >= self.OMISSIONS_LEVEL_1 and m >= self.MOMENTUM_LEVEL_1:
level = 3 # S
elif o >= self.OMISSIONS_LEVEL_2 and m >= self.MOMENTUM_LEVEL_2:
level = 2 # A
elif o >= self.OMISSIONS_LEVEL_3 and m >= self.MOMENTUM_LEVEL_3:
level = 1 # B
else:
level = 0
if level > 0:
scores.append({'num': n, 'level': level, 'o': o, 'm': m})
if not scores:
return 0, []
max_level = max(s['level'] for s in scores)
if max_level == 3:
max_bets = 8
elif max_level == 2:
max_bets = 6
else:
max_bets = 4
scores.sort(key=lambda x: (x['level'], x['o'], x['m']), reverse=True)
top_nums = [s['num'] for s in scores[:max_bets]]
return len(top_nums), top_nums
def _check_position_elite(self, pos, history_data):
"""检查位置是否为精英位置 (在 predict 中简化为 True)"""
return True
def check_init_daily_pnl_tracker(self, current_date):
_date_range = None
for date_range in self.daily_pnl_tracker:
start_time, end_time = date_range
if start_time <= current_date <= end_time:
_date_range = (start_time, end_time)
if not _date_range:
current_date_start_time = pd.to_datetime(current_date.date()) + timedelta(hours=7, minutes=5)
current_date_end_time = current_date_start_time + timedelta(hours=22, minutes=55)
_date_range = (current_date_start_time, current_date_end_time)
self.daily_pnl_tracker[_date_range] = {'pnl': 0, 'bets': 0, 'miss_count': 0}
return _date_range
def _check_daily_stop_loss(self, current_date):
"""检查是否触发日内止损"""
stats = self.daily_pnl_tracker[self.check_init_daily_pnl_tracker(current_date)]
if not stats:
raise Exception(f"{current_date} not in {list(self.daily_pnl_tracker.keys())}")
# 先不算水钱
net_pnl = stats['pnl']
# net_pnl = stats['pnl'] + stats['bets'] * self.REBATE_RATE
if stats['miss_count'] >= 2:
# 连续 miss 取 小值
# print(f"行情不好, 少投点: {stats['miss_count']}")
self._bet_this_round = self._bet_this_round if self._bet_this_round < (self.BASE_BET_UNIT * 0.1) else self.BASE_BET_UNIT * 0.1
if net_pnl < -(self.INITIAL_CAPITAL * self.DAILY_STOP_LOSS_RATE):
# print("止损 =========")
return True
return False
def update_result(self, period_record, bets_made):
"""
在实际开奖后调用，用于更新日内盈亏追踪器
:param period_record: 包含 'result' 和 'time' 的记录
:param bets_made: 实际投注建议 {position: [num1, num2, ...]}
"""
if not bets_made:
return
current_date = period_record["time"].iloc[0]
# current_date = period_record["time"].dt.strftime("%Y-%m-%d").iloc[0]
# current_date = datetime.strptime(period_record['time'], '%Y-%m-%d %H:%M:%S').date()
pnl_this_period = 0
bets_this_period = 0
for pos, nums in bets_made["result"].items():
predict_nums = []
bets = 0
for predict_num, bet in nums.items():
if bet:
predict_nums.append(predict_num)
bets += bet
hit = list(period_record['result'])[0][pos] in predict_nums
pnl = (self._bet_this_round * self.ODDS if hit else 0) - bets
pnl_this_period += pnl
bets_this_period += bets
date_range = self.check_init_daily_pnl_tracker(current_date)
self.daily_pnl_tracker[date_range]['pnl'] += pnl_this_period
self.daily_pnl_tracker[date_range]['bets'] += bets_this_period
if pnl_this_period < bets_this_period:
self.daily_pnl_tracker[date_range]['miss_count'] += 1
else:
self.daily_pnl_tracker[date_range]['miss_count'] = 0
def _get_current_actual_pnl(self, current_date):
day_data = self.daily_pnl_tracker[self.check_init_daily_pnl_tracker(current_date)]
return day_data['pnl']
# return day_data['pnl'] + (day_data['bets'] * self.REBATE_RATE)
def _set_bet_level(self, actual_pnl):
if actual_pnl == 0:
return self.BASE_BET_UNIT * 0.1
if actual_pnl < 0:
if actual_pnl < -(self.SAFE_BET_LEVEL_4 * self.INITIAL_CAPITAL):
return self.BASE_BET_UNIT * 0.6
if actual_pnl < -(self.SAFE_BET_LEVEL_3 * self.INITIAL_CAPITAL):
return self.BASE_BET_UNIT * 0.4
if actual_pnl < -(self.SAFE_BET_LEVEL_2 * self.INITIAL_CAPITAL):
return self.BASE_BET_UNIT * 0.3
if actual_pnl < -(self.SAFE_BET_LEVEL_1 * self.INITIAL_CAPITAL):
return self.BASE_BET_UNIT * 0.2
return self.BASE_BET_UNIT
else:
if actual_pnl > self.RADICAL_BET_LEVEL_2 * self.INITIAL_CAPITAL:
return self.BASE_BET_UNIT * 1.2
if actual_pnl > self.RADICAL_BET_LEVEL_1 * self.INITIAL_CAPITAL:
return self.BASE_BET_UNIT * 1.5
if actual_pnl > self.RADICAL_BET_LEVEL_0 * self.INITIAL_CAPITAL:
return self.BASE_BET_UNIT * 2
return self.BASE_BET_UNIT
def predict(self, current_date, history_data):
"""
核心预测方法
:param current_period_record: 当前期的记录 (用于获取时间)
:param history_data: 历史数据列表
:return: 投注建议字典 {position: [num1, num2, ...]}
"""
# 1. 动态金额调节
actual_pnl = self._get_current_actual_pnl(current_date)
self._bet_this_round = self._set_bet_level(actual_pnl)
# 2. 日内止损检查
if self._check_daily_stop_loss(current_date):
return {}
omissions, momentum = self._calculate_omission_and_momentum(history_data)
if omissions is None:
return {}
bet_suggestions = {}
for pos in range(10):
# 2. 动态位置筛选
# elite_pos = self._get_elite_positions(history_df)
# if not elite_pos:
# continue
# 3. 动态注数投注
num_bets, top_nums = self._get_dynamic_bet_nums(pos, omissions, momentum)
if num_bets > 0:
bet_suggestions[pos] = {num: self._bet_this_round for num in top_nums}
# if self._check_daily_stop_loss(current_date):
# for pos, num_bets in bet_suggestions.items():
# bet_suggestions[pos] = {
# num: self.BASE_BET_UNIT * self.RADICAL_BET_LEVEL_0 for num, _ in num_bets.items()
# }
return bet_suggestions