paper-dynasty-database/app/db_engine.py
Cal Corum 4bfd878486 feat: add PlayerSeasonStats Peewee model (#67)
Closes #67

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-12 16:35:02 -05:00

1397 lines
38 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import math
from datetime import datetime
from typing import List
import logging
import os
from pandas import DataFrame
from peewee import *
from peewee import ModelSelect
from playhouse.shortcuts import model_to_dict
# Database configuration - supports both SQLite and PostgreSQL
DATABASE_TYPE = os.environ.get("DATABASE_TYPE", "sqlite")
# Skip table creation for PostgreSQL (tables already exist in production)
SKIP_TABLE_CREATION = DATABASE_TYPE.lower() == "postgresql"
if DATABASE_TYPE.lower() == "postgresql":
from playhouse.pool import PooledPostgresqlDatabase
db = PooledPostgresqlDatabase(
os.environ.get("POSTGRES_DB", "pd_master"),
user=os.environ.get("POSTGRES_USER", "pd_admin"),
password=os.environ.get("POSTGRES_PASSWORD"),
host=os.environ.get("POSTGRES_HOST", "localhost"),
port=int(os.environ.get("POSTGRES_PORT", "5432")),
max_connections=20,
stale_timeout=300, # 5 minutes
timeout=0,
autoconnect=True,
autorollback=True, # Automatically rollback failed transactions
)
else:
# SQLite configuration for local development only.
# Production always uses PostgreSQL (see DATABASE_TYPE env var).
#
# synchronous=0 (OFF): SQLite skips fsync() after every write, maximising
# throughput at the cost of durability — a hard crash could corrupt the DB.
# This is an acceptable trade-off in dev where data loss is tolerable and
# write speed matters. WAL journal mode reduces (but does not eliminate)
# the corruption window by keeping the main database file consistent while
# writes land in the WAL file first.
db = SqliteDatabase(
"storage/pd_master.db",
pragmas={"journal_mode": "wal", "cache_size": -1 * 64000, "synchronous": 0},
)
# 2025, 2005
ranked_cardsets = [24, 25, 26, 27, 28, 29]
LIVE_CARDSET_ID = 27
LIVE_PROMO_CARDSET_ID = 28
CARDSETS = {
"ranked": {"primary": ranked_cardsets, "human": ranked_cardsets},
"minor-league": {
"primary": [27, 8], # 2005, Mario
"secondary": [24], # 2025
"human": [x for x in range(1, 30)],
},
"major-league": {
"primary": [
27,
28,
24,
25,
13,
14,
6,
8,
], # 2005 + Promos, 2025 + Promos, 2018 + Promos, 2012, Mario
"secondary": [5, 3], # 2019, 2022
"human": ranked_cardsets,
},
"hall-of-fame": {"primary": [x for x in range(1, 30)], "human": ranked_cardsets},
"flashback": {
"primary": [
13,
14,
5,
1,
3,
4,
8,
], # 2018 + Promos, 2019, 2021, 2022 + Promos, Mario
"secondary": [24], # 2025
"human": [13, 14, 5, 1, 3, 4, 8], # 2018 + Promos, 2019, 2021, 2022 + Promos
},
"gauntlet-3": {
"primary": [13], # 2018
"secondary": [5, 11, 9], # 2019, 2016, 2023
"human": [x for x in range(1, 30)],
},
"gauntlet-4": {
"primary": [3, 6, 16], # 2022, 2013, Backyard Baseball
"secondary": [4, 9], # 2022 Promos, 2023
"human": [3, 4, 6, 9, 15, 16],
},
"gauntlet-5": {
"primary": [17, 8], # 2024, Mario
"secondary": [13], # 2018
"human": [x for x in range(1, 30)],
},
"gauntlet-6": {
"primary": [20, 8], # 1998, Mario
"secondary": [12], # 2008
"human": [x for x in range(1, 30)],
},
"gauntlet-7": {
"primary": [5, 23], # 2019, Brilliant Stars
"secondary": [1], # 2021
"human": [x for x in range(1, 30)],
},
"gauntlet-8": {
"primary": [24], # 2025
"secondary": [17],
"human": [24, 25, 22, 23],
},
"gauntlet-9": {
"primary": [27], # 2005
"secondary": [24], # 2025
"human": [x for x in range(1, 30)],
},
}
def model_csv_headers(this_obj, exclude=None) -> List:
data = model_to_dict(this_obj, recurse=False, exclude=exclude)
return [x for x in data.keys()]
def model_to_csv(this_obj, exclude=None) -> List:
data = model_to_dict(this_obj, recurse=False, exclude=exclude)
return [x for x in data.values()]
def query_to_csv(all_items: ModelSelect, exclude=None):
if all_items.count() == 0:
data_list = [["No data found"]]
else:
data_list = [model_csv_headers(all_items[0], exclude=exclude)]
for x in all_items:
data_list.append(model_to_csv(x, exclude=exclude))
return DataFrame(data_list).to_csv(header=False, index=False)
def complex_data_to_csv(complex_data: List):
if len(complex_data) == 0:
data_list = [["No data found"]]
else:
data_list = [[x for x in complex_data[0].keys()]]
for line in complex_data:
logging.debug(f"line: {line}")
this_row = []
for key in line:
logging.debug(f"key: {key}")
if line[key] is None:
this_row.append("")
elif isinstance(line[key], dict):
if "name" in line[key]:
this_row.append(line[key]["name"])
elif "abbrev" in line[key]:
this_row.append(line[key]["abbrev"])
else:
this_row.append(line[key]["id"])
elif isinstance(line[key], int) and line[key] > 100000000:
this_row.append(f"'{line[key]}")
elif isinstance(line[key], str) and "," in line[key]:
this_row.append(line[key].replace(",", "-_-"))
else:
this_row.append(line[key])
data_list.append(this_row)
return DataFrame(data_list).to_csv(header=False, index=False)
class BaseModel(Model):
class Meta:
database = db
class Current(BaseModel):
season = IntegerField()
week = IntegerField(default=0)
gsheet_template = CharField()
gsheet_version = CharField()
live_scoreboard = BigIntegerField() # Discord channel ID - needs BIGINT
class Meta:
database = db
table_name = "current"
@staticmethod
def latest():
latest_current = Current.select().order_by(-Current.id).get()
return latest_current
if not SKIP_TABLE_CREATION:
db.create_tables([Current], safe=True)
class Rarity(BaseModel):
value = IntegerField()
name = CharField(unique=True)
color = CharField()
class Meta:
database = db
table_name = "rarity"
def __str__(self):
return self.name
if not SKIP_TABLE_CREATION:
db.create_tables([Rarity], safe=True)
class Event(BaseModel):
name = CharField()
short_desc = CharField(null=True)
url = CharField(null=True)
long_desc = CharField(null=True)
thumbnail = CharField(null=True)
active = BooleanField(default=False)
class Meta:
database = db
table_name = "event"
if not SKIP_TABLE_CREATION:
db.create_tables([Event], safe=True)
class Cardset(BaseModel):
name = CharField()
description = CharField()
event = ForeignKeyField(Event, null=True)
for_purchase = BooleanField(default=True) # for_purchase
total_cards = IntegerField()
in_packs = BooleanField(default=True)
ranked_legal = BooleanField(default=True)
class Meta:
database = db
table_name = "cardset"
def __str__(self):
return self.name
if not SKIP_TABLE_CREATION:
db.create_tables([Cardset], safe=True)
class MlbPlayer(BaseModel):
first_name = CharField()
last_name = CharField()
key_fangraphs = IntegerField(null=True)
key_bbref = CharField(null=True)
key_retro = CharField(null=True)
key_mlbam = IntegerField(null=True)
offense_col = IntegerField(default=1)
class Meta:
database = db
table_name = "mlbplayer"
if not SKIP_TABLE_CREATION:
db.create_tables([MlbPlayer], safe=True)
class Player(BaseModel):
player_id = IntegerField(primary_key=True)
p_name = CharField()
cost = IntegerField(default=0)
image = CharField()
image2 = CharField(null=True)
mlbclub = CharField()
franchise = CharField()
cardset = ForeignKeyField(Cardset)
set_num = IntegerField()
rarity = ForeignKeyField(Rarity)
pos_1 = CharField()
pos_2 = CharField(null=True)
pos_3 = CharField(null=True)
pos_4 = CharField(null=True)
pos_5 = CharField(null=True)
pos_6 = CharField(null=True)
pos_7 = CharField(null=True)
pos_8 = CharField(null=True)
headshot = CharField(null=True)
vanity_card = CharField(null=True)
strat_code = CharField(null=True)
bbref_id = CharField(null=True)
fangr_id = CharField(null=True)
description = CharField()
quantity = IntegerField(default=999)
mlbplayer = ForeignKeyField(MlbPlayer, null=True)
def __str__(self):
return f"{self.cardset} {self.p_name} ({self.rarity.name})"
# def __eq__(self, other):
# if self.cardset.id == other.cardset.id and self.name == other.name:
# return True
# else:
# return False
def __lt__(self, other):
if self.wara < other.wara:
return True
elif self.wara > other.wara:
return False
elif self.name < other.name:
return True
else:
return False
def get_all_pos(self):
all_pos = []
if self.pos_1 and self.pos_1 != "CP":
all_pos.append(self.pos_1)
if self.pos_2 and self.pos_2 != "CP":
all_pos.append(self.pos_2)
if self.pos_3 and self.pos_3 != "CP":
all_pos.append(self.pos_3)
if self.pos_4 and self.pos_4 != "CP":
all_pos.append(self.pos_4)
if self.pos_5 and self.pos_5 != "CP":
all_pos.append(self.pos_5)
if self.pos_6 and self.pos_6 != "CP":
all_pos.append(self.pos_6)
if self.pos_7 and self.pos_7 != "CP":
all_pos.append(self.pos_7)
if self.pos_8 and self.pos_8 != "CP":
all_pos.append(self.pos_8)
return all_pos
def change_on_sell(self):
# caps = {
# 'replacement': 15,
# 'reserve': 50,
# 'starter': 200,
# 'all-star': 750,
# 'mvp': 2500,
# 'hof': 999999999
# }
logging.info(f"{self.p_name} cost changing from: {self.cost}")
self.cost = max(math.floor(self.cost * 0.95), 1)
# if self.quantity != 999:
# self.quantity += 1
logging.info(f"{self.p_name} cost now: {self.cost}")
self.save()
def change_on_buy(self):
logging.info(f"{self.p_name} cost changing from: {self.cost}")
self.cost = math.ceil(self.cost * 1.1)
# if self.quantity != 999:
# self.quantity -= 1
logging.info(f"{self.p_name} cost now: {self.cost}")
self.save()
class Meta:
database = db
table_name = "player"
if not SKIP_TABLE_CREATION:
db.create_tables([Player], safe=True)
class Team(BaseModel):
abbrev = CharField(max_length=20) # Gauntlet teams use prefixes like "Gauntlet-NCB"
sname = CharField(max_length=100)
lname = CharField(max_length=255)
gmid = BigIntegerField() # Discord user ID - needs BIGINT
gmname = CharField()
gsheet = CharField()
wallet = IntegerField()
team_value = IntegerField()
collection_value = IntegerField()
logo = CharField(null=True)
color = CharField(null=True)
season = IntegerField()
event = ForeignKeyField(Event, null=True)
career = IntegerField(default=0)
ranking = IntegerField(default=1000)
has_guide = BooleanField(default=False)
is_ai = IntegerField(null=True)
def __str__(self):
return f"S{self.season} {self.lname}"
@staticmethod
def get_by_owner(gmid, season=None):
if not season:
season = Current.get().season
team = Team.get_or_none((Team.gmid == gmid) & (Team.season == season))
if not team:
return None
return team
@staticmethod
def select_season(season=None):
if not season:
season = Current.get().season
return Team.select().where(Team.season == season)
@staticmethod
def get_season(abbrev, season=None):
if not season:
season = Current.get().season
return Team.get_or_none(Team.season == season, Team.abbrev == abbrev.upper())
def team_hash(self):
hash_string = f"{self.sname[-1]}{self.gmid / 6950123:.0f}{self.sname[-2]}{self.gmid / 42069123:.0f}"
logging.info(f"string: {hash_string}")
return hash_string
class Meta:
database = db
table_name = "team"
if not SKIP_TABLE_CREATION:
db.create_tables([Team], safe=True)
class PackType(BaseModel):
name = CharField()
card_count = IntegerField()
description = CharField()
cost = IntegerField()
available = BooleanField(default=True)
class Meta:
database = db
table_name = "packtype"
if not SKIP_TABLE_CREATION:
db.create_tables([PackType], safe=True)
class Pack(BaseModel):
team = ForeignKeyField(Team)
pack_type = ForeignKeyField(PackType)
pack_team = ForeignKeyField(Team, null=True)
pack_cardset = ForeignKeyField(Cardset, null=True)
open_time = DateTimeField(null=True)
class Meta:
database = db
table_name = "pack"
if not SKIP_TABLE_CREATION:
db.create_tables([Pack], safe=True)
class Card(BaseModel):
player = ForeignKeyField(Player, null=True)
team = ForeignKeyField(Team, null=True)
pack = ForeignKeyField(Pack, null=True)
value = IntegerField(default=0)
def __str__(self):
if self.player:
return f"{self.player} - {self.team.sname}"
else:
return f"Blank - {self.team.sname}"
@staticmethod
def select_season(season):
return Card.select().join(Team).where(Card.team.season == season)
class Meta:
database = db
table_name = "card"
if not SKIP_TABLE_CREATION:
db.create_tables([Card], safe=True)
class Roster(BaseModel):
team = ForeignKeyField(Team)
name = CharField()
roster_num = IntegerField()
def __str__(self):
return f"{self.team} Roster"
def get_cards(self):
return (
Card.select()
.join(RosterSlot)
.where(RosterSlot.roster == self)
.order_by(RosterSlot.slot)
)
class Meta:
database = db
table_name = "roster"
class RosterSlot(BaseModel):
roster = ForeignKeyField(Roster, backref="slots")
slot = IntegerField()
card = ForeignKeyField(Card, backref="roster_slots")
class Meta:
database = db
table_name = "rosterslot"
indexes = ((("roster", "slot"), True),)
class Result(BaseModel):
away_team = ForeignKeyField(Team)
home_team = ForeignKeyField(Team)
away_score = IntegerField()
home_score = IntegerField()
away_team_value = IntegerField(null=True)
home_team_value = IntegerField(null=True)
away_team_ranking = IntegerField(null=True)
home_team_ranking = IntegerField(null=True)
scorecard = CharField()
week = IntegerField()
season = IntegerField()
ranked = BooleanField()
short_game = BooleanField()
game_type = CharField(null=True)
@staticmethod
def select_season(season=None):
if not season:
season = Current.get().season
return Result.select().where(Result.season == season)
class Meta:
database = db
table_name = "result"
class BattingStat(BaseModel):
card = ForeignKeyField(Card)
team = ForeignKeyField(Team)
roster_num = IntegerField()
vs_team = ForeignKeyField(Team)
result = ForeignKeyField(Result, null=True)
pos = CharField()
pa = IntegerField()
ab = IntegerField()
run = IntegerField()
hit = IntegerField()
rbi = IntegerField()
double = IntegerField()
triple = IntegerField()
hr = IntegerField()
bb = IntegerField()
so = IntegerField()
hbp = IntegerField()
sac = IntegerField()
ibb = IntegerField()
gidp = IntegerField()
sb = IntegerField()
cs = IntegerField()
bphr = IntegerField()
bpfo = IntegerField()
bp1b = IntegerField()
bplo = IntegerField()
xch = IntegerField()
xhit = IntegerField()
error = IntegerField()
pb = IntegerField()
sbc = IntegerField()
csc = IntegerField()
week = IntegerField()
season = IntegerField()
created = DateTimeField()
game_id = IntegerField()
class Meta:
database = db
table_name = "battingstat"
class PitchingStat(BaseModel):
card = ForeignKeyField(Card)
team = ForeignKeyField(Team)
roster_num = IntegerField()
vs_team = ForeignKeyField(Team)
result = ForeignKeyField(Result, null=True)
ip = FloatField()
hit = IntegerField()
run = IntegerField()
erun = IntegerField()
so = IntegerField()
bb = IntegerField()
hbp = IntegerField()
wp = IntegerField()
balk = IntegerField()
hr = IntegerField()
ir = IntegerField()
irs = IntegerField()
gs = IntegerField()
win = IntegerField()
loss = IntegerField()
hold = IntegerField()
sv = IntegerField()
bsv = IntegerField()
week = IntegerField()
season = IntegerField()
created = DateTimeField()
game_id = IntegerField()
class Meta:
database = db
table_name = "pitchingstat"
class Award(BaseModel):
name = CharField()
season = IntegerField()
timing = CharField(default="In-Season")
card = ForeignKeyField(Card, null=True)
team = ForeignKeyField(Team, null=True)
image = CharField(null=True)
class Meta:
database = db
table_name = "award"
class Paperdex(BaseModel):
team = ForeignKeyField(Team)
player = ForeignKeyField(Player)
created = DateTimeField(default=datetime.now)
class Meta:
database = db
table_name = "paperdex"
# def add_to_paperdex(self, team, cards: list):
# for x in players:
# if not isinstance(x, Card):
# raise TypeError(f'The Pokedex can only take a list of Player or Card objects')
#
# Paperdex.get_or_create(team=team, player=player)
class Reward(BaseModel):
name = CharField(null=True)
season = IntegerField()
week = IntegerField()
team = ForeignKeyField(Team)
created = DateTimeField()
class Meta:
database = db
table_name = "reward"
class GameRewards(BaseModel):
name = CharField()
pack_type = ForeignKeyField(PackType, null=True)
player = ForeignKeyField(Player, null=True)
money = IntegerField(null=True)
class Meta:
database = db
table_name = "gamerewards"
class Notification(BaseModel):
created = DateTimeField()
title = CharField()
desc = CharField(null=True)
field_name = CharField()
message = CharField()
about = CharField() # f'{Topic}-{Object ID}'
ack = BooleanField(default=False)
class Meta:
database = db
table_name = "notification"
class GauntletReward(BaseModel):
name = CharField()
gauntlet = ForeignKeyField(Event)
reward = ForeignKeyField(GameRewards)
win_num = IntegerField()
loss_max = IntegerField(default=1)
class Meta:
database = db
table_name = "gauntletreward"
class GauntletRun(BaseModel):
team = ForeignKeyField(Team)
gauntlet = ForeignKeyField(Event)
wins = IntegerField(default=0)
losses = IntegerField(default=0)
gsheet = CharField(null=True)
created = DateTimeField(default=datetime.now)
ended = DateTimeField(null=True) # NULL means run not yet ended
class Meta:
database = db
table_name = "gauntletrun"
if not SKIP_TABLE_CREATION:
db.create_tables(
[
Roster,
RosterSlot,
BattingStat,
PitchingStat,
Result,
Award,
Paperdex,
Reward,
GameRewards,
Notification,
GauntletReward,
GauntletRun,
],
safe=True,
)
class BattingCard(BaseModel):
player = ForeignKeyField(Player)
variant = IntegerField()
steal_low = IntegerField()
steal_high = IntegerField()
steal_auto = BooleanField()
steal_jump = FloatField()
bunting = CharField()
hit_and_run = CharField()
running = IntegerField()
offense_col = IntegerField()
hand = CharField(default="R")
class Meta:
database = db
table_name = "battingcard"
bc_index = ModelIndex(
BattingCard, (BattingCard.player, BattingCard.variant), unique=True
)
BattingCard.add_index(bc_index)
class BattingCardRatings(BaseModel):
battingcard = ForeignKeyField(BattingCard)
vs_hand = CharField(default="R")
pull_rate = FloatField()
center_rate = FloatField()
slap_rate = FloatField()
homerun = FloatField()
bp_homerun = FloatField()
triple = FloatField()
double_three = FloatField()
double_two = FloatField()
double_pull = FloatField()
single_two = FloatField()
single_one = FloatField()
single_center = FloatField()
bp_single = FloatField()
hbp = FloatField()
walk = FloatField()
strikeout = FloatField()
lineout = FloatField()
popout = FloatField()
flyout_a = FloatField()
flyout_bq = FloatField()
flyout_lf_b = FloatField()
flyout_rf_b = FloatField()
groundout_a = FloatField()
groundout_b = FloatField()
groundout_c = FloatField()
avg = FloatField(null=True)
obp = FloatField(null=True)
slg = FloatField(null=True)
class Meta:
database = db
table_name = "battingcardratings"
bcr_index = ModelIndex(
BattingCardRatings,
(BattingCardRatings.battingcard, BattingCardRatings.vs_hand),
unique=True,
)
BattingCardRatings.add_index(bcr_index)
class PitchingCard(BaseModel):
player = ForeignKeyField(Player)
variant = IntegerField()
balk = IntegerField()
wild_pitch = IntegerField()
hold = IntegerField()
starter_rating = IntegerField()
relief_rating = IntegerField()
closer_rating = IntegerField(null=True)
batting = CharField(null=True)
offense_col = IntegerField()
hand = CharField(default="R")
class Meta:
database = db
table_name = "pitchingcard"
pc_index = ModelIndex(
PitchingCard, (PitchingCard.player, PitchingCard.variant), unique=True
)
PitchingCard.add_index(pc_index)
class PitchingCardRatings(BaseModel):
pitchingcard = ForeignKeyField(PitchingCard)
vs_hand = CharField(default="R")
homerun = FloatField()
bp_homerun = FloatField()
triple = FloatField()
double_three = FloatField()
double_two = FloatField()
double_cf = FloatField()
single_two = FloatField()
single_one = FloatField()
single_center = FloatField()
bp_single = FloatField()
hbp = FloatField()
walk = FloatField()
strikeout = FloatField()
flyout_lf_b = FloatField()
flyout_cf_b = FloatField()
flyout_rf_b = FloatField()
groundout_a = FloatField()
groundout_b = FloatField()
xcheck_p = FloatField()
xcheck_c = FloatField()
xcheck_1b = FloatField()
xcheck_2b = FloatField()
xcheck_3b = FloatField()
xcheck_ss = FloatField()
xcheck_lf = FloatField()
xcheck_cf = FloatField()
xcheck_rf = FloatField()
avg = FloatField(null=True)
obp = FloatField(null=True)
slg = FloatField(null=True)
class Meta:
database = db
table_name = "pitchingcardratings"
pcr_index = ModelIndex(
PitchingCardRatings,
(PitchingCardRatings.pitchingcard, PitchingCardRatings.vs_hand),
unique=True,
)
PitchingCardRatings.add_index(pcr_index)
class CardPosition(BaseModel):
player = ForeignKeyField(Player)
variant = IntegerField()
position = CharField()
innings = IntegerField()
range = IntegerField()
error = IntegerField()
arm = IntegerField(null=True)
pb = IntegerField(null=True)
overthrow = IntegerField(null=True)
class Meta:
database = db
table_name = "cardposition"
pos_index = ModelIndex(
CardPosition,
(CardPosition.player, CardPosition.variant, CardPosition.position),
unique=True,
)
CardPosition.add_index(pos_index)
if not SKIP_TABLE_CREATION:
db.create_tables(
[
BattingCard,
BattingCardRatings,
PitchingCard,
PitchingCardRatings,
CardPosition,
],
safe=True,
)
class StratGame(BaseModel):
season = IntegerField()
game_type = CharField()
away_team = ForeignKeyField(Team)
home_team = ForeignKeyField(Team)
week = IntegerField(default=1)
away_score = IntegerField(default=0)
home_score = IntegerField(default=0)
away_team_value = IntegerField(null=True)
home_team_value = IntegerField(null=True)
away_team_ranking = IntegerField(null=True)
home_team_ranking = IntegerField(null=True)
ranked = BooleanField(default=False)
short_game = BooleanField(default=False)
forfeit = BooleanField(default=False)
class Meta:
database = db
table_name = "stratgame"
class StratPlay(BaseModel):
game = ForeignKeyField(StratGame)
play_num = IntegerField()
batter = ForeignKeyField(Player, null=True)
batter_team = ForeignKeyField(Team, null=True)
pitcher = ForeignKeyField(Player)
pitcher_team = ForeignKeyField(Team)
on_base_code = CharField()
inning_half = CharField()
inning_num = IntegerField()
batting_order = IntegerField()
starting_outs = IntegerField()
away_score = IntegerField()
home_score = IntegerField()
batter_pos = CharField(null=True)
# These <base>_final fields track the base this runner advances to post-play (None) if out
on_first = ForeignKeyField(Player, null=True)
on_first_final = IntegerField(null=True)
on_second = ForeignKeyField(Player, null=True)
on_second_final = IntegerField(null=True)
on_third = ForeignKeyField(Player, null=True)
on_third_final = IntegerField(null=True)
batter_final = IntegerField(null=True)
pa = IntegerField(default=0)
ab = IntegerField(default=0)
e_run = IntegerField(default=0)
run = IntegerField(default=0)
hit = IntegerField(default=0)
rbi = IntegerField(default=0)
double = IntegerField(default=0)
triple = IntegerField(default=0)
homerun = IntegerField(default=0)
bb = IntegerField(default=0)
so = IntegerField(default=0)
hbp = IntegerField(default=0)
sac = IntegerField(default=0)
ibb = IntegerField(default=0)
gidp = IntegerField(default=0)
bphr = IntegerField(default=0)
bpfo = IntegerField(default=0)
bp1b = IntegerField(default=0)
bplo = IntegerField(default=0)
sb = IntegerField(default=0)
cs = IntegerField(default=0)
outs = IntegerField(default=0)
wpa = FloatField(default=0.0)
re24 = FloatField(default=0.0)
# These <position> fields are only required if the play is an x-check or baserunning play
catcher = ForeignKeyField(Player, null=True)
catcher_team = ForeignKeyField(Team, null=True)
defender = ForeignKeyField(Player, null=True)
defender_team = ForeignKeyField(Team, null=True)
runner = ForeignKeyField(Player, null=True)
runner_team = ForeignKeyField(Team, null=True)
check_pos = CharField(null=True)
error = IntegerField(default=0)
wild_pitch = IntegerField(default=0)
passed_ball = IntegerField(default=0)
pick_off = IntegerField(default=0)
balk = IntegerField(default=0)
is_go_ahead = BooleanField(default=False)
is_tied = BooleanField(default=False)
is_new_inning = BooleanField(default=False)
class Meta:
database = db
table_name = "stratplay"
# Unique index for StratPlay - a play number should be unique within a game
# Required for PostgreSQL on_conflict() upsert operations
stratplay_index = ModelIndex(
StratPlay, (StratPlay.game, StratPlay.play_num), unique=True
)
StratPlay.add_index(stratplay_index)
class Decision(BaseModel):
season = IntegerField()
game = ForeignKeyField(StratGame)
pitcher = ForeignKeyField(Player)
pitcher_team = ForeignKeyField(Team)
week = IntegerField(default=1)
win = IntegerField(default=0)
loss = IntegerField(default=0)
hold = IntegerField(default=0)
is_save = IntegerField(default=0)
b_save = IntegerField(default=0)
irunners = IntegerField(default=0)
irunners_scored = IntegerField(default=0)
rest_ip = FloatField(default=0.0)
rest_required = IntegerField(default=0)
is_start = BooleanField(default=False)
class Meta:
database = db
table_name = "decision"
# Unique index for Decision - one decision per pitcher per game
# Required for PostgreSQL on_conflict() upsert operations
decision_index = ModelIndex(Decision, (Decision.game, Decision.pitcher), unique=True)
Decision.add_index(decision_index)
class PlayerSeasonStats(BaseModel):
player = ForeignKeyField(Player)
team = ForeignKeyField(Team)
season = IntegerField()
# Batting stats
games_batting = IntegerField(default=0)
pa = IntegerField(default=0)
ab = IntegerField(default=0)
hits = IntegerField(default=0)
hr = IntegerField(default=0)
doubles = IntegerField(default=0)
triples = IntegerField(default=0)
bb = IntegerField(default=0)
hbp = IntegerField(default=0)
so = IntegerField(default=0)
rbi = IntegerField(default=0)
runs = IntegerField(default=0)
sb = IntegerField(default=0)
cs = IntegerField(default=0)
# Pitching stats
games_pitching = IntegerField(default=0)
outs = IntegerField(default=0)
k = IntegerField(
default=0
) # pitcher Ks; spec names this "so (K)" but renamed to avoid collision with batting so
bb_allowed = IntegerField(default=0)
hits_allowed = IntegerField(default=0)
hr_allowed = IntegerField(default=0)
wins = IntegerField(default=0)
losses = IntegerField(default=0)
saves = IntegerField(default=0)
holds = IntegerField(default=0)
blown_saves = IntegerField(default=0)
# Meta
last_game = ForeignKeyField(StratGame, null=True)
last_updated_at = DateTimeField(null=True)
class Meta:
database = db
table_name = "player_season_stats"
pss_unique_index = ModelIndex(
PlayerSeasonStats,
(PlayerSeasonStats.player, PlayerSeasonStats.team, PlayerSeasonStats.season),
unique=True,
)
PlayerSeasonStats.add_index(pss_unique_index)
pss_team_season_index = ModelIndex(
PlayerSeasonStats,
(PlayerSeasonStats.team, PlayerSeasonStats.season),
unique=False,
)
PlayerSeasonStats.add_index(pss_team_season_index)
pss_player_season_index = ModelIndex(
PlayerSeasonStats,
(PlayerSeasonStats.player, PlayerSeasonStats.season),
unique=False,
)
PlayerSeasonStats.add_index(pss_player_season_index)
if not SKIP_TABLE_CREATION:
db.create_tables([StratGame, StratPlay, Decision, PlayerSeasonStats], safe=True)
class ScoutOpportunity(BaseModel):
pack = ForeignKeyField(Pack, null=True)
opener_team = ForeignKeyField(Team)
card_ids = CharField() # JSON array of card IDs
expires_at = BigIntegerField()
created = BigIntegerField()
class Meta:
database = db
table_name = "scout_opportunity"
class ScoutClaim(BaseModel):
scout_opportunity = ForeignKeyField(ScoutOpportunity)
card = ForeignKeyField(Card)
claimed_by_team = ForeignKeyField(Team)
created = BigIntegerField()
class Meta:
database = db
table_name = "scout_claim"
scout_claim_index = ModelIndex(
ScoutClaim,
(ScoutClaim.scout_opportunity, ScoutClaim.claimed_by_team),
unique=True,
)
ScoutClaim.add_index(scout_claim_index)
if not SKIP_TABLE_CREATION:
db.create_tables([ScoutOpportunity, ScoutClaim], safe=True)
class EvolutionTrack(BaseModel):
name = CharField()
card_type = CharField() # batter / sp / rp
formula = CharField()
t1_threshold = IntegerField()
t2_threshold = IntegerField()
t3_threshold = IntegerField()
t4_threshold = IntegerField()
class Meta:
database = db
table_name = "evolution_track"
class EvolutionCardState(BaseModel):
player = ForeignKeyField(Player)
team = ForeignKeyField(Team)
track = ForeignKeyField(EvolutionTrack)
current_tier = IntegerField(default=0) # valid range: 04
current_value = FloatField(default=0.0)
fully_evolved = BooleanField(default=False)
last_evaluated_at = DateTimeField(null=True)
class Meta:
database = db
table_name = "evolution_card_state"
ecs_index = ModelIndex(
EvolutionCardState,
(EvolutionCardState.player, EvolutionCardState.team),
unique=True,
)
EvolutionCardState.add_index(ecs_index)
class EvolutionTierBoost(BaseModel):
"""Phase 2 stub — minimal model, schema to be defined in phase 2."""
card_state = ForeignKeyField(EvolutionCardState)
class Meta:
database = db
table_name = "evolution_tier_boost"
class EvolutionCosmetic(BaseModel):
"""Phase 2 stub — minimal model, schema to be defined in phase 2."""
card_state = ForeignKeyField(EvolutionCardState)
class Meta:
database = db
table_name = "evolution_cosmetic"
if not SKIP_TABLE_CREATION:
db.create_tables(
[EvolutionTrack, EvolutionCardState, EvolutionTierBoost, EvolutionCosmetic],
safe=True,
)
db.close()
# scout_db = SqliteDatabase(
# 'storage/card_creation.db',
# pragmas={
# 'journal_mode': 'wal',
# 'cache_size': -1 * 64000,
# 'synchronous': 0
# }
# )
#
#
# class BaseModelScout(Model):
# class Meta:
# database = scout_db
#
#
# class ScoutCardset(BaseModelScout):
# set_title = CharField()
# set_subtitle = CharField(null=True)
#
#
# class ScoutPlayer(BaseModelScout):
# sba_id = IntegerField(primary_key=True)
# name = CharField()
# fg_id = IntegerField()
# br_id = CharField()
# offense_col = IntegerField()
# hand = CharField(default='R')
#
#
# scout_db.create_tables([ScoutCardset, ScoutPlayer], safe=True)
#
#
# class BatterRatings(BaseModelScout):
# id = CharField(unique=True, primary_key=True)
# player = ForeignKeyField(ScoutPlayer)
# cardset = ForeignKeyField(ScoutCardset)
# vs_hand = FloatField()
# is_prep = BooleanField()
# homerun = FloatField()
# bp_homerun = FloatField()
# triple = FloatField()
# double_three = FloatField()
# double_two = FloatField()
# double_pull = FloatField()
# single_two = FloatField()
# single_one = FloatField()
# single_center = FloatField()
# bp_single = FloatField()
# hbp = FloatField()
# walk = FloatField()
# strikeout = FloatField()
# lineout = FloatField()
# popout = FloatField()
# flyout_a = FloatField()
# flyout_bq = FloatField()
# flyout_lf_b = FloatField()
# flyout_rf_b = FloatField()
# groundout_a = FloatField()
# groundout_b = FloatField()
# groundout_c = FloatField()
# avg = FloatField(null=True)
# obp = FloatField(null=True)
# slg = FloatField(null=True)
#
#
# class PitcherRatings(BaseModelScout):
# id = CharField(unique=True, primary_key=True)
# player = ForeignKeyField(ScoutPlayer)
# cardset = ForeignKeyField(ScoutCardset)
# vs_hand = CharField()
# is_prep = BooleanField()
# homerun = FloatField()
# bp_homerun = FloatField()
# triple = FloatField()
# double_three = FloatField()
# double_two = FloatField()
# double_cf = FloatField()
# single_two = FloatField()
# single_one = FloatField()
# single_center = FloatField()
# bp_single = FloatField()
# hbp = FloatField()
# walk = FloatField()
# strikeout = FloatField()
# fo_slap = FloatField()
# fo_center = FloatField()
# groundout_a = FloatField()
# groundout_b = FloatField()
# xcheck_p = FloatField()
# xcheck_c = FloatField()
# xcheck_1b = FloatField()
# xcheck_2b = FloatField()
# xcheck_3b = FloatField()
# xcheck_ss = FloatField()
# xcheck_lf = FloatField()
# xcheck_cf = FloatField()
# xcheck_rf = FloatField()
# avg = FloatField(null=True)
# obp = FloatField(null=True)
# slg = FloatField(null=True)
#
#
# # scout_db.create_tables([BatterRatings, PitcherRatings], safe=True)
#
#
# class CardColumns(BaseModelScout):
# id = CharField(unique=True, primary_key=True)
# player = ForeignKeyField(ScoutPlayer)
# hand = CharField()
# b_ratings = ForeignKeyField(BatterRatings, null=True)
# p_ratings = ForeignKeyField(PitcherRatings, null=True)
# one_dice = CharField()
# one_results = CharField()
# one_splits = CharField()
# two_dice = CharField()
# two_results = CharField()
# two_splits = CharField()
# three_dice = CharField()
# three_results = CharField()
# three_splits = CharField()
#
#
# class Position(BaseModelScout):
# player = ForeignKeyField(ScoutPlayer)
# cardset = ForeignKeyField(ScoutCardset)
# position = CharField()
# innings = IntegerField()
# range = IntegerField()
# error = IntegerField()
# arm = CharField(null=True)
# pb = IntegerField(null=True)
# overthrow = IntegerField(null=True)
#
#
# class BatterData(BaseModelScout):
# player = ForeignKeyField(ScoutPlayer)
# cardset = ForeignKeyField(ScoutCardset)
# stealing = CharField()
# st_low = IntegerField()
# st_high = IntegerField()
# st_auto = BooleanField()
# st_jump = FloatField()
# bunting = CharField(null=True)
# hit_and_run = CharField(null=True)
# running = CharField()
#
#
# class PitcherData(BaseModelScout):
# player = ForeignKeyField(ScoutPlayer)
# cardset = ForeignKeyField(ScoutCardset)
# balk = IntegerField(null=True)
# wild_pitch = IntegerField(null=True)
# hold = CharField()
# starter_rating = IntegerField()
# relief_rating = IntegerField()
# closer_rating = IntegerField(null=True)
# batting = CharField(null=True)
#
#
# scout_db.create_tables([CardColumns, Position, BatterData, PitcherData], safe=True)
#
#
# class CardOutput(BaseModelScout):
# name = CharField()
# hand = CharField()
# positions = CharField()
# stealing = CharField()
# bunting = CharField()
# hitandrun = CharField()
# running = CharField()
#
#
# scout_db.close()