移除main

BetMetices.py

@@ -0,0 +1,66 @@
+import pandas as pd
+from common.bet_tools import compute_metrics
+from common.bet_tools import calculate_no_vig_moneyline_power, moneyline_to_prob
+from loguru import logger
+from common.utils import timeit
+
+
+def get_no_vig_prob(row) -> pd.Series:
+    odds = [row["first_price"], row["second_price"]]
+    no_vig_odds_power = calculate_no_vig_moneyline_power(odds)
+    novig_probs_power = [moneyline_to_prob(o) for o in no_vig_odds_power]
+    # 返回两个无水概率
+    return pd.Series(
+        {
+            "first_no_vig_prob": novig_probs_power[0],
+            "second_no_vig_prob": novig_probs_power[1],
+        }
+    )
+
+
+@timeit
+def calc_metrics(df: pd.DataFrame, cols: list) -> pd.DataFrame:
+    data_list = []
+    for cs in df[cols].drop_duplicates().values:
+        tmp_df = df[cols + ["win_prob", "res"]].copy()
+        for i, col in enumerate(cols):
+            tmp_df = tmp_df[tmp_df[col] == cs[i]]
+        # if len(tmp_df) < 10000:
+        #     continue
+        res = compute_metrics(df=tmp_df, include_draws=False)
+        res["filter_cols"] = ",".join(cs)
+        data_list.append(res)
+    res_df = pd.DataFrame(data_list)
+    res_df["reg_alpha"] = abs(res_df["reg_alpha"])
+    res_df = res_df.sort_values(by=["brier", "logloss", "ece", "reg_alpha"])
+    return res_df
+
+
+if __name__ == "__main__":
+    df = pd.read_csv(
+        "/Users/aszer/Documents/vscode/bet/data/pinnical_1xbet_all_api.csv",
+        encoding="utf-8-sig",
+    )
+    df = df[
+        [
+            "sportsbook",
+            "sport",
+            "league",
+            "fixture_id",
+            "game_id",
+            "market",
+            "first_price",
+            "second_price",
+            "market_width",
+            "result",
+        ]
+    ]
+    # 防止 SettingWithCopyWarning，推荐使用 .loc 显式分配
+    df.loc[:, ["first_no_vig_prob", "second_no_vig_prob"]] = df[
+        ["first_price", "second_price"]
+    ].apply(get_no_vig_prob, axis=1)
+    df["win_prob"] = df["first_no_vig_prob"]
+    df["res"] = df["result"]
+    cols = ["sportsbook", "sport"]
+    res_df = calc_metrics(df, cols)
+    logger.info(f"\n{res_df}")

main.py

@@ -1,59 +0,0 @@
-
-import pandas as pd
-from typing import Generator
-from pytz import timezone
-from dao.Database import Database
-from data_model import MysqlConfig, OddsJamOrder, OddsjamBet
-
-
-def generate_oddjams_bet(data: OddsJamOrder) -> Generator[OddsjamBet]:
-    # 根据 start_timestamp 是否小于当前时间两个小时判断要不要去看结果
-    # 怎么判断一个 oddjam order 是不是下单成功？
-    bet_name = data.home_bet_name
-    sport_books = data.home_sportsbooks
-    if data.away_price > data.away_no_vig_price:
-        bet_name = data.away_bet_name
-        sport_books = data.away_sportsbooks # 选 selected_sportsbook
-
-    # 时区转换, mysql 中的 game start date 是 utc 时间？ 这个不需要
-
-    game_start_date = data.start_date
-    my_datetime_with_tz = game_start_date.replace(tzinfo=timezone('UTC'))
-    eastern = timezone('US/Eastern')
-    eastern_datetime = my_datetime_with_tz.astimezone(eastern)
-    game_start_date = eastern_datetime.strftime('%m/%d/%Y, %H:%M EDT')
-
-    for sport_book in sport_books:
-        yield OddsjamBet(
-            Sportsbook=sport_book,
-            BetName=bet_name,
-            MarketName=data.market,
-            Odds=110, # away price or home price
-            Stake=1,
-            EventName='', # home team vs away team 用这种格式， 需要带vs和两边的空格
-            Sport=data.sport,
-            League=data.league,
-            GameID=data.game_id,
-            GameStartDate=game_start_date,
-            BetType='Positive EV',
-            Notes=data.bet_id
-        )
-
-
-if __name__ == '__main__':
-    config_file_path = 'config\mysql_config.json'
-    mysql_config = MysqlConfig.parse_file(config_file_path)
-    dao = Database(mysql_config)
-
-    # query
-    select_query = "SELECT * FROM bet.oddsjam_order limit 10;"
-    raw_data_list = dao.fetchall(query=select_query)
-    order_data_list = [OddsJamOrder(**data) for data in raw_data_list]
-
-    bet_list = []
-    for order in order_data_list:
-        bet_list.extend(list(generate_oddjams_bet(order)))
-
-    bet_list = [data.model_dump() for data in bet_list]
-    bet_df = pd.DataFrame(bet_list)
-    bet_df.to_csv('bet.csv', index=False, encoding='utf-8-sig')