sasheg/dodep-simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
47c7f17c9036e2cf703c62da4aae82834c6e9ead
dodep-simulator/rpu.py
T
sasheg 47c7f17c90 RPU v2, promo/bank система, админка, оптимизации 
— RPU v2: бюджет удачи, потолок (ceiling), комбек после
  проигрышей, hot/cold анализ, интеграция с promo-фазами
— Bank API: mock-банк для карты (депозит/вывод с балансом 1000₽)
— Promo-сервис: автогенерация промокодов каждый час, типы
  (card_to_site, luck_boost, ceiling_boost, free_case, reset_streak)
— Promo-фазы: LUCKY/NEUTRAL/UNLUCKY, переключение каждые 1-6ч
— Admin: история транзакций (user_history), управление промокодами,
  управление promo-фазами, API фильтров/коллекций, RPU info
— Database: WAL mode, новые поля RPU v2, TransactionLog,
  карточный баланс, статистики (streaks, ceiling, budget)
— Frontend: /deposit, /promo/activate, /withdraw, /card-balance,
  /online-count, /rpu-info, lifespan вместо startup event
— UI: user_history.html, улучшения профиля, кейсов, навигации
— Cases: обновление cases.json (1892 строки новых данных)
— Achievements: кэш коллекций, оптимизация is_custom проверок
— Sounds: дополнительные звуковые эффекты
— Upgrade: интеграция с RPU (множители, streaks)
— .gitignore: исключены .db, .bak, __pycache__
2026-07-25 13:11:08 +00:00

685 lines
23 KiB
Python
Raw Blame History

import json
import random
import math
from datetime import datetime, date
from typing import Dict, Optional, Tuple
from sqlalchemy.orm import Session
from database import User, UserRPU, InventoryItem
from promo_phase import get_promo_effect_for_user, get_current_phase
# Стандартные веса редкостей
BASE_RARITY_WEIGHTS = {
"Consumer Grade": 10000,
"Industrial Grade": 5000,
"Mil-Spec": 2500,
"Mil-Spec Grade": 2500,
"Restricted": 625,
"Classified": 125,
"Covert": 25,
"Rare Special Item": 5,
"Extraordinary": 5
}
RARITY_TO_FIELD = {
"Consumer Grade": "consumer_multiplier",
"Industrial Grade": "industrial_multiplier",
"Mil-Spec": "mil_spec_multiplier",
"Mil-Spec Grade": "mil_spec_multiplier",
"Restricted": "restricted_multiplier",
"Classified": "classified_multiplier",
"Covert": "covert_multiplier",
"Rare Special Item": "rare_special_multiplier",
"Extraordinary": "rare_special_multiplier"
}
RARITY_RANK = {
"Consumer Grade": 0,
"Industrial Grade": 1,
"Mil-Spec": 2,
"Mil-Spec Grade": 2,
"Restricted": 3,
"Classified": 4,
"Covert": 5,
"Rare Special Item": 6,
"Extraordinary": 6
}
# Пороги потолка
CEILING_BASE_MULTIPLIER = 3.0
CEILING_MIN = 1000.0
CEILING_SESSION_HISTORY = {
0: 5.0, # первая сессия
1: 4.0, # вторая
2: 3.5, # третья
-1: 3.0 # 3+ сессий
}
# Пороги для luck_budget
BUDGET_MAX = 100.0
BUDGET_CONSUMPTION_RATE = 0.001 # 0.1% от цены предмета
BUDGET_REGEN_PER_OPEN = 0.5 # +0.5% за каждое открытие
BUDGET_LOW_THRESHOLD = 20.0 # если budget < 20%, удача падает
# Комбек
COMEBACK_TRIGGER_COUNT = 5 # 5+ проигрышей подряд → комбек
COMEBACK_TRIGGER_VALUE_RATIO = 0.3 # потеря >30% от total_deposited
COMEBACK_OPENINGS = 5 # комбек на 5 открытий
COMEBACK_MULTIPLIER_MAX = 3.0 # макс множитель удачи в комбеке
# Hot/Cold
HOT_THRESHOLD_SPENT = 10000 # потратил >10k
HOT_THRESHOLD_WITHDRAWAL = 0.1 # вывел <10% от депозитов
COLD_DAYS_INACTIVE = 7 # 7 дней неактивен = холодный
COLD_SPENT_PER_DAY = 100 # <100₽/день = холодный
def get_user_rpu(db: Session, user_id: int) -> UserRPU:
rpu = db.query(UserRPU).filter(UserRPU.user_id == user_id).first()
if not rpu:
rpu = UserRPU(user_id=user_id)
db.add(rpu)
_rpu_commit_or_flush(db)
db.refresh(rpu)
return rpu
# ── Session Management ──
def check_and_reset_session(db: Session, user_id: int):
"""
Проверяет, не пора ли сбросить сессию (каждый календарный день).
"""
rpu = get_user_rpu(db, user_id)
today = date.today()
if rpu.session_reset_date is None:
rpu.session_reset_date = datetime.utcnow()
rpu.luck_budget = BUDGET_MAX
rpu.session_spent = 0
rpu.session_won = 0
rpu.ceiling_break_session = 0
rpu.comeback_active = False
rpu.comeback_openings_left = 0
rpu.comeback_multiplier = 1.0
rpu.consecutive_loss_count = 0
rpu.consecutive_loss_value = 0.0
_rpu_commit_or_flush(db)
return
last_reset = rpu.session_reset_date.date()
if last_reset < today:
rpu.session_reset_date = datetime.utcnow()
rpu.luck_budget = BUDGET_MAX
rpu.session_spent = 0
rpu.session_won = 0
rpu.ceiling_break_session = 0
rpu.comeback_active = False
rpu.comeback_openings_left = 0
rpu.comeback_multiplier = 1.0
rpu.consecutive_loss_count = 0
rpu.consecutive_loss_value = 0.0
_rpu_commit_or_flush(db)
# ── Ceiling (Потолок) ──
def calculate_ceiling(db: Session, user_id: int) -> float:
"""
Потолок = total_deposited * mult.
mult зависит от количества сессий на сайте.
"""
user = db.query(User).filter(User.id == user_id).first()
if not user:
return CEILING_MIN
total_dep = max(0.0, user.total_deposited)
# Считаем, сколько прошло сессий (дней с первой транзакции)
rpu = get_user_rpu(db, user_id)
session_count = 0
if rpu.session_reset_date and user.created_at:
days_since = (datetime.utcnow() - user.created_at).days
session_count = max(0, days_since)
if session_count == 0:
mult = CEILING_SESSION_HISTORY[0]
elif session_count == 1:
mult = CEILING_SESSION_HISTORY[1]
elif session_count == 2:
mult = CEILING_SESSION_HISTORY[2]
else:
mult = CEILING_SESSION_HISTORY[-1]
# Учитываем ceiling_multiplier (временные модификаторы)
mult *= rpu.ceiling_multiplier
ceiling = total_dep * mult
return max(CEILING_MIN, ceiling)
def get_total_inventory_value(db: Session, user_id: int) -> float:
"""Суммарная стоимость инвентаря пользователя"""
items = db.query(InventoryItem).filter(InventoryItem.user_id == user_id).all()
return sum(item.price_rub or 0 for item in items)
def get_user_total_value(db: Session, user_id: int) -> float:
"""Баланс + инвентарь"""
user = db.query(User).filter(User.id == user_id).first()
balance = user.balance if user else 0
inv = get_total_inventory_value(db, user_id)
return balance + inv
def get_ceiling_luck_factor(db: Session, user_id: int) -> float:
"""
Возвращает множитель удачи на основе близости к потолку (0.3–1.0).
Чем ближе к потолку — тем меньше множитель.
Не блокирует открытие, а лишь делает удачу чуть хуже.
"""
user = db.query(User).filter(User.id == user_id).first()
if not user:
return 1.0
ceiling = calculate_ceiling(db, user_id)
inv_value = get_total_inventory_value(db, user_id)
withdrawn = user.total_withdrawn or 0
extracted_value = inv_value + withdrawn
if ceiling <= 0:
return 1.0
ratio = extracted_value / ceiling
if ratio < 0.8:
return 1.0
if ratio >= 1.0:
return 0.3
return 1.0 - (ratio - 0.8) / 0.2 * 0.7
def check_ceiling_breach(db: Session, user_id: int, item_price: float,
case_price: float) -> bool:
"""
Проверяет, пробивает ли предмет потолок.
Если предмет дороже 50% от потолка — пробитие.
"""
user = db.query(User).filter(User.id == user_id).first()
if not user:
return False
ceiling = calculate_ceiling(db, user_id)
total_value = user.balance + get_total_inventory_value(db, user_id)
# Если инвентарь уже под потолком — не пробиваем
if total_value < ceiling:
return False
# Сам предмет должен быть дорогим
if item_price < ceiling * 0.5:
return False
return True
def apply_ceiling_breach(db: Session, user_id: int, item_price: float):
"""
Пробитие потолка: расширяем потолок на 50% на остаток сессии.
"""
rpu = get_user_rpu(db, user_id)
rpu.ceiling_multiplier = min(3.0, rpu.ceiling_multiplier * 1.5)
rpu.ceiling_break_count += 1
rpu.ceiling_break_session += 1
_rpu_commit_or_flush(db)
def apply_ceiling_withdrawal_penalty(db: Session, user_id: int):
"""
При большом выводе — скручиваем потолок на следующих N сессий.
"""
rpu = get_user_rpu(db, user_id)
rpu.ceiling_multiplier = max(0.5, rpu.ceiling_multiplier * 0.7)
_rpu_commit_or_flush(db)
# ── Luck Budget ──
# Флаг: если True — sub-функции делают flush вместо commit.
# Устанавливается в True на время горячего пути (case_open, upgrade).
_BATCH_MODE = False
# Кэш adjusted_weights на время batch-операции
# Ключ: user_id, значение: результат get_adjusted_weights
_BATCH_WEIGHTS_CACHE = {}
def _rpu_commit_or_flush(db):
if _BATCH_MODE:
db.flush()
else:
db.commit()
def _batch_cache_get(user_id: int):
"""Возвращает кэшированные adjusted_weights для batch или None."""
if _BATCH_MODE and user_id in _BATCH_WEIGHTS_CACHE:
return _BATCH_WEIGHTS_CACHE[user_id]
return None
def _batch_cache_set(user_id: int, weights: dict):
"""Сохраняет adjusted_weights в кэш batch."""
if _BATCH_MODE:
_BATCH_WEIGHTS_CACHE[user_id] = weights
def _batch_cache_clear(user_id: int = None):
"""Очищает кэш batch (по user_id или полностью)."""
if user_id is not None:
_BATCH_WEIGHTS_CACHE.pop(user_id, None)
else:
_BATCH_WEIGHTS_CACHE.clear()
def consume_luck_budget(db: Session, user_id: int, item_price: float):
rpu = get_user_rpu(db, user_id)
consumption = item_price * BUDGET_CONSUMPTION_RATE
rpu.luck_budget = max(0.0, rpu.luck_budget - consumption)
_rpu_commit_or_flush(db)
def regen_luck_budget(db: Session, user_id: int):
rpu = get_user_rpu(db, user_id)
rpu.luck_budget = min(BUDGET_MAX, rpu.luck_budget + BUDGET_REGEN_PER_OPEN)
_rpu_commit_or_flush(db)
def get_budget_factor(rpu: UserRPU) -> float:
"""
Возвращает множитель на основе luck_budget.
Если budget низкий (<20%) — удача падает.
"""
if rpu.luck_budget >= BUDGET_LOW_THRESHOLD:
return 1.0
return max(0.3, rpu.luck_budget / BUDGET_LOW_THRESHOLD)
# ── Comeback System ──
def update_comeback_tracking(db: Session, user_id: int, spent: float,
received_value: float):
"""
Отслеживает проигрыши для триггера комбека.
"""
rpu = get_user_rpu(db, user_id)
user = db.query(User).filter(User.id == user_id).first()
if not user:
return
# Если проиграл (получил меньше чем потратил)
if received_value < spent * 0.8:
rpu.consecutive_loss_count += 1
rpu.consecutive_loss_value += (spent - received_value)
else:
rpu.consecutive_loss_count = 0
rpu.consecutive_loss_value = 0.0
# Проверяем триггер комбека
total_dep = max(1.0, user.total_deposited)
loss_ratio = rpu.consecutive_loss_value / total_dep
if (rpu.consecutive_loss_count >= COMEBACK_TRIGGER_COUNT or
loss_ratio >= COMEBACK_TRIGGER_VALUE_RATIO):
if not rpu.comeback_active:
rpu.comeback_active = True
rpu.comeback_openings_left = COMEBACK_OPENINGS
rpu.comeback_multiplier = COMEBACK_MULTIPLIER_MAX
_rpu_commit_or_flush(db)
def apply_comeback(db: Session, user_id: int) -> float:
"""
Применяет комбек-множитель если активен.
Возвращает множитель для следующего открытия.
"""
rpu = get_user_rpu(db, user_id)
if not rpu.comeback_active or rpu.comeback_openings_left <= 0:
return 1.0
mult = rpu.comeback_multiplier
rpu.comeback_openings_left -= 1
if rpu.comeback_openings_left <= 0:
rpu.comeback_active = False
rpu.comeback_multiplier = 1.0
rpu.consecutive_loss_count = 0
rpu.consecutive_loss_value = 0.0
_rpu_commit_or_flush(db)
return mult
# ── Hot/Cold Analysis ──
def analyze_hot_cold(db: Session, user_id: int) -> Tuple[bool, bool, float]:
"""
Анализирует пользователя: горячий / холодный.
Возвращает (is_hot, is_cold, withdrawal_ratio).
"""
user = db.query(User).filter(User.id == user_id).first()
if not user:
return False, True, 0.0
total_dep = max(0.0, user.total_deposited)
total_wd = max(0.0, user.total_withdrawn)
wd_ratio = total_wd / max(1.0, total_dep)
# Горячий: много потратил, почти не выводил
is_hot = (total_dep >= HOT_THRESHOLD_SPENT and wd_ratio < HOT_THRESHOLD_WITHDRAWAL)
# Холодный: давно не был или мало активность
rpu = get_user_rpu(db, user_id)
is_cold = False
if rpu.last_activity_date:
days_since = (datetime.utcnow() - rpu.last_activity_date).days
if days_since >= COLD_DAYS_INACTIVE:
is_cold = True
else:
is_cold = True # Новичок без активности
# Считаем hot_score
days_active = max(1, (datetime.utcnow() - user.created_at).days)
spent_per_day = total_dep / days_active
if spent_per_day < COLD_SPENT_PER_DAY and total_dep > 0:
is_cold = True
hot_score = total_dep - (total_wd * 2)
rpu.hot_score = hot_score
rpu.last_activity_date = datetime.utcnow()
_rpu_commit_or_flush(db)
return is_hot, is_cold, wd_ratio
# ── Withdrawal Tax (Открутка при выводе) ──
def check_withdrawal_penalty(db: Session, user_id: int):
"""
Если пользователь вывел >50% от депозитов — скручиваем удачу и потолок.
"""
user = db.query(User).filter(User.id == user_id).first()
if not user:
return
total_dep = max(1.0, user.total_deposited)
total_wd = max(0.0, user.total_withdrawn)
wd_ratio = total_wd / total_dep
if wd_ratio > 0.5:
rpu = get_user_rpu(db, user_id)
rpu.luck_multiplier = max(0.3, rpu.luck_multiplier - wd_ratio * 0.3)
rpu.ceiling_multiplier = max(0.5, rpu.ceiling_multiplier * 0.7)
_rpu_commit_or_flush(db)
# ── V2: Enhanced get_adjusted_weights ──
def get_adjusted_weights(db: Session, user_id: int) -> Dict[str, float]:
"""
Получает веса редкостей с учетом РПУ v2:
- сессии, потолок, luck_budget, комбек, промо-фаза, hot/cold
В batch-режиме результат кэшируется — повторные вызовы с тем же user_id
возвращают закешированное значение без DB-запросов.
"""
cached = _batch_cache_get(user_id)
if cached is not None:
return cached
check_and_reset_session(db, user_id)
rpu = get_user_rpu(db, user_id)
# Промо-фаза
promo = get_promo_effect_for_user(user_id, rpu.hot_score, wd_ratio, is_cold)
# Комбек
comeback_mult = apply_comeback(db, user_id)
# Budget factor
budget_factor = get_budget_factor(rpu)
# Ceiling factor — не блокирует, а лишь приглушает удачу
ceiling_factor = get_ceiling_luck_factor(db, user_id)
adjusted = {}
for rarity, base_weight in BASE_RARITY_WEIGHTS.items():
field = RARITY_TO_FIELD.get(rarity)
base_multiplier = getattr(rpu, field, 1.0) if field else 1.0
rarity_index = RARITY_RANK.get(rarity, 0)
if rpu_level < 50:
luck_factor = 1.0 + ((50 - rpu_level) / 50) * rarity_index * 0.3
else:
luck_factor = 1.0 - ((rpu_level - 50) / 50) * rarity_index * 0.2
luck_factor = max(0.1, min(3.0, luck_factor))
# Применяем все множители
final = base_multiplier * rpu.luck_multiplier * luck_factor
# Промо-фаза
final *= (1.0 + promo["luck_modifier"])
# Budget
final *= budget_factor
# Комбек
final *= comeback_mult
# Потолок — не блокирует, а лишь приглушает удачу
final *= ceiling_factor
# Редкие предметы получают доп модификатор от промо
if rarity_index >= 3:
final *= (1.0 + promo.get("rare_chance_modifier", 0))
final = max(0.05, min(5.0, final))
adjusted[rarity] = base_weight * final
_batch_cache_set(user_id, adjusted)
return adjusted
def calculate_rpu_level(rpu: UserRPU) -> float:
multipliers = [
rpu.consumer_multiplier,
rpu.industrial_multiplier,
rpu.mil_spec_multiplier,
rpu.restricted_multiplier,
rpu.classified_multiplier,
rpu.covert_multiplier,
rpu.rare_special_multiplier
]
avg = sum(multipliers) / len(multipliers)
if avg >= 1.0:
level = 50 - ((avg - 1.0) / 2.0) * 100
else:
level = 50 + ((1.0 - avg) / 0.9) * 50
return max(0, min(100, level))
def get_rpu_description(level: float) -> str:
if level < 20:
return "🍀 Очень везучий"
elif level < 35:
return "😊 Везучий"
elif level < 45:
return "🙂 Немного везучий"
elif level < 55:
return "😐 Обычный"
elif level < 70:
return "🙁 Немного невезучий"
elif level < 85:
return "😟 Невезучий"
else:
return "💀 Проклятый"
def auto_adjust_rpu(db: Session, user_id: int):
rpu = get_user_rpu(db, user_id)
if not rpu.auto_adjust:
return
user = db.query(User).filter(User.id == user_id).first()
if not user:
return
if rpu.total_opened > 0 and rpu.total_spent > 0:
roi = rpu.total_value_received / rpu.total_spent
if roi < 0.4:
boost = (0.4 - roi) * 2
rpu.luck_multiplier = min(2.5, rpu.luck_multiplier + boost)
for r in ['restricted_multiplier', 'classified_multiplier', 'covert_multiplier', 'rare_special_multiplier']:
v = getattr(rpu, r)
setattr(rpu, r, min(2.0, v + boost * 0.3))
elif roi > 2.0:
penalty = (roi - 2.0) * 0.5
rpu.luck_multiplier = max(0.3, rpu.luck_multiplier - penalty)
for r in ['restricted_multiplier', 'classified_multiplier', 'covert_multiplier', 'rare_special_multiplier']:
v = getattr(rpu, r)
setattr(rpu, r, max(0.3, v - penalty * 0.5))
streak = rpu.current_streak
if streak <= -5:
boost = min(0.3, abs(streak) * 0.03)
rpu.luck_multiplier = min(2.5, rpu.luck_multiplier + boost)
rpu.classified_multiplier = min(2.0, rpu.classified_multiplier + boost * 0.5)
rpu.covert_multiplier = min(1.8, rpu.covert_multiplier + boost * 0.4)
rpu.rare_special_multiplier = min(1.5, rpu.rare_special_multiplier + boost * 0.3)
if streak >= 5:
penalty = min(0.2, streak * 0.02)
rpu.luck_multiplier = max(0.3, rpu.luck_multiplier - penalty)
for r in ['classified_multiplier', 'covert_multiplier', 'rare_special_multiplier']:
v = getattr(rpu, r)
setattr(rpu, r, max(0.2, v - penalty * 0.5))
if rpu.total_opened > 50:
rpu.classified_multiplier = min(1.5, rpu.classified_multiplier + 0.05)
rpu.covert_multiplier = min(1.3, rpu.covert_multiplier + 0.03)
for field in ['consumer_multiplier', 'industrial_multiplier', 'mil_spec_multiplier',
'restricted_multiplier', 'classified_multiplier', 'covert_multiplier',
'rare_special_multiplier']:
value = getattr(rpu, field)
setattr(rpu, field, max(0.1, min(3.0, value)))
rpu.luck_multiplier = max(0.1, min(3.0, rpu.luck_multiplier))
rpu.last_adjustment = datetime.utcnow()
_rpu_commit_or_flush(db)
def update_rpu_stats(db: Session, user_id: int, spent: float, opened: int = 1,
received_value: float = 0.0, rarity: str = None,
is_rare: bool = False):
rpu = get_user_rpu(db, user_id)
user = db.query(User).filter(User.id == user_id).first()
rpu.total_spent += spent
rpu.total_opened += opened
rpu.total_value_received += received_value
rpu.session_spent += spent
rpu.session_won += received_value
if is_rare or received_value > spent * 3:
rpu.current_streak = min(20, rpu.current_streak + 1)
else:
rpu.current_streak = max(-20, rpu.current_streak - 1)
if rpu.current_streak > rpu.best_streak:
rpu.best_streak = rpu.current_streak
if rpu.current_streak < rpu.worst_streak:
rpu.worst_streak = rpu.current_streak
# ── RPU v2: Luck Budget ──
consume_luck_budget(db, user_id, received_value)
regen_luck_budget(db, user_id)
# ── RPU v2: Comeback tracking ──
update_comeback_tracking(db, user_id, spent, received_value)
# ── RPU v2: Ceiling breach check ──
if user and is_rare and received_value > 0:
if check_ceiling_breach(db, user_id, received_value, spent):
apply_ceiling_breach(db, user_id, received_value)
# ── RPU v2: Hot/Cold ──
analyze_hot_cold(db, user_id)
_rpu_commit_or_flush(db)
if rpu.auto_adjust:
auto_adjust_rpu(db, user_id)
def get_rpu_v2_info(db: Session, user_id: int) -> dict:
"""
Возвращает полную информацию о RPU v2 для профиля/админки.
"""
check_and_reset_session(db, user_id)
rpu = get_user_rpu(db, user_id)
user = db.query(User).filter(User.id == user_id).first()
if not user:
return {}
ceiling = calculate_ceiling(db, user_id)
inv_value = get_total_inventory_value(db, user_id)
total_value = user.balance + inv_value
is_hot, is_cold, wd_ratio = analyze_hot_cold(db, user_id)
session_count = 0
if user.created_at:
session_count = max(0, (datetime.utcnow() - user.created_at).days)
return {
"rpu_level": round(calculate_rpu_level(rpu)),
"description": get_rpu_description(calculate_rpu_level(rpu)),
"luck_multiplier": rpu.luck_multiplier,
"luck_budget": round(rpu.luck_budget, 1),
"ceiling": round(ceiling, 2),
"ceiling_multiplier": round(rpu.ceiling_multiplier, 2),
"ceiling_breach_total": rpu.ceiling_break_count,
"ceiling_breach_session": rpu.ceiling_break_session,
"total_value": round(total_value, 2),
"balance": round(user.balance, 2),
"inventory_value": round(inv_value, 2),
"total_deposited": round(user.total_deposited, 2),
"total_withdrawn": round(user.total_withdrawn, 2),
"session": {
"number": session_count,
"spent": round(rpu.session_spent, 2),
"won": round(rpu.session_won, 2),
},
"comeback": {
"active": rpu.comeback_active,
"openings_left": rpu.comeback_openings_left,
"multiplier": round(rpu.comeback_multiplier, 2),
},
"streak": {
"current": rpu.current_streak,
"best": rpu.best_streak,
"worst": rpu.worst_streak,
},
"hot_cold": {
"is_hot": is_hot,
"is_cold": is_cold,
"hot_score": round(rpu.hot_score, 2),
"withdrawal_ratio": round(wd_ratio, 3),
},
"promo_phase": get_current_phase(),
}
Reference in New Issue View Git Blame Copy Permalink