libs.domain.aggregate

  1"""
  2libs/domain/aggregate.py
  3"""
  4
  5from typing import Any, Optional
  6
  7import numpy as np
  8import pandas as pd
  9
 10import libs.global_value as g
 11from libs.utils import formatter
 12
 13
 14def game_summary(
 15    filter_items: Optional[list[str]] = None,
 16    drop_items: Optional[list[str]] = None,
 17) -> pd.DataFrame:
 18    """
 19    ゲーム結果をサマライズする
 20
 21    Args:
 22        filter_items (Optional[list[str]]): 抽出するカラム. Defaults to None.
 23        drop_items (Optional[list[str]]): 除外するカラム. Defaults to None.
 24
 25    Returns:
 26        pd.DataFrame: 集計結果
 27
 28    """
 29    # データ収集
 30    df = g.params.read_data("SUMMARY_TOTAL")
 31
 32    # 順位分布選択
 33    match g.params.mode:
 34        case 3:
 35            df = df.drop(columns=["rank_distr4"])
 36        case 4:
 37            df = df.drop(columns=["rank_distr3"])
 38
 39    if isinstance(filter_items, list):
 40        df = df.filter(items=filter_items)
 41
 42    if isinstance(drop_items, list):
 43        df = df.drop(columns=drop_items)
 44
 45    return df
 46
 47
 48def calculation_rating() -> pd.DataFrame:
 49    """
 50    レーティング集計
 51
 52    Returns:
 53        pd.DataFrame: 集計結果
 54
 55    """
 56    # データ収集
 57    df_results = g.params.read_data("RANKING_RATINGS").set_index("playtime")
 58    df_ratings = pd.DataFrame(index=["initial_rating"] + df_results.index.to_list())  # 記録用
 59    last_ratings: dict[str, float] = {}  # 最終値格納用
 60
 61    # 獲得スコア
 62    score_mapping = {"1": 30.0, "2": 10.0, "3": -10.0, "4": -30.0}
 63
 64    for x in df_results.itertuples():
 65        player_list = (str(x.p1_name), str(x.p2_name), str(x.p3_name), str(x.p4_name))
 66        for player in player_list:
 67            if player not in df_ratings.columns:
 68                last_ratings[player] = 1500.0
 69                df_ratings[player] = np.nan
 70                df_ratings.loc["initial_rating", player] = 1500.0
 71                df_ratings = df_ratings
 72
 73        # 天鳳計算式 (https://tenhou.net/man/#RATING)
 74        rank_list = (x.p1_rank, x.p2_rank, x.p3_rank, x.p4_rank)
 75        rating_list = [last_ratings[player] for player in player_list]
 76        rating_avg = float(1500.0 if np.mean(rating_list) < 1500.0 else np.mean(rating_list))
 77
 78        for i, player in enumerate(player_list):
 79            rating = float(rating_list[i])
 80            rank = str(rank_list[i])
 81
 82            correction_value: float = (rating_avg - rating) / 40
 83            if df_ratings[player].count() >= 400:
 84                match_correction = 0.2
 85            else:
 86                match_correction = 1 - df_ratings[player].count() * 0.002
 87
 88            new_rating = rating + match_correction * (score_mapping[rank] + correction_value)
 89
 90            last_ratings[player] = new_rating
 91            df_ratings.loc[x.Index, player] = new_rating
 92
 93    # 間引き(集約オプション)
 94    if collection := g.params.collection:
 95        ratings = df_ratings[1:]
 96        ratings.index = pd.to_datetime(ratings.index)  # DatetimeIndexに変換
 97
 98        match collection:
 99            case "daily":
100                ratings = ratings.resample("D").last().ffill()
101            case "monthly":
102                ratings = ratings.resample("ME").last().ffill()
103            case "yearly":
104                ratings = ratings.resample("YE").last().ffill()
105            case "all":
106                ratings = df_ratings.ffill().tail(1)
107            case _:
108                return df_ratings
109
110        ratings.index = ratings.index.astype(str)
111        df_ratings = pd.concat([df_ratings.head(1), ratings])
112
113    return df_ratings
114
115
116def grade_promotion_check(grade_level: int, point: int, rank: int) -> tuple[int, int]:
117    """
118    昇段チェック
119
120    Args:
121        grade_level (int): 現在のレベル(段位)
122        point (int): 現在の昇段ポイント
123        rank (int): 獲得順位
124
125    Returns:
126        tuple[int, int]: チェック後の昇段ポイント, チェック後のレベル(段位)
127
128    """
129    tbl_data = g.cfg.badge.grade.table["table"]
130    new_point = point + int(tbl_data[grade_level]["acquisition"][rank - 1])
131
132    if new_point >= int(tbl_data[grade_level]["point"][1]):  # level up
133        grade_level = min(grade_level + 1, len(tbl_data) - 1)
134        new_point = int(tbl_data[grade_level]["point"][0])  # 初期値
135    elif new_point < 0:  # level down
136        new_point = int(0)
137        if tbl_data[grade_level]["demote"]:
138            grade_level = max(grade_level - 1, 0)
139            new_point = int(tbl_data[grade_level]["point"][0])  # 初期値
140
141    return (new_point, grade_level)
142
143
144# レポート
145def matrix_table() -> pd.DataFrame:
146    """
147    対局対戦マトリックス表の作成
148
149    Returns:
150        pd.DataFrame: 集計結果
151
152    """
153    # データ収集
154    df = g.params.read_data("REPORT_MATRIX_TABLE").set_index("playtime")
155
156    # 結果に含まれるプレイヤーのリスト
157    plist = sorted(list(set(df["p1_name"].tolist() + df["p2_name"].tolist() + df["p3_name"].tolist() + df["p4_name"].tolist())))
158
159    # 順位テーブルの作成
160    l_data: dict[str, Any] = {}
161    for pname in plist:
162        if g.params.individual:  # 個人集計
163            l_name = formatter.name_replace(pname)
164            # プレイヤー指定があるなら対象以外をスキップ
165            if g.params.player_list:
166                if l_name not in g.params.player_list:
167                    continue
168            # ゲスト置換
169            if g.params.guest_skip:  # ゲストあり
170                l_name = formatter.name_replace(pname, add_mark=True)
171            else:  # ゲストなし
172                if pname == g.cfg.member.guest_name:
173                    continue
174        else:  # チーム集計
175            l_name = pname
176
177        l_data[l_name] = []
178        for x in df.itertuples():
179            match pname:
180                case x.p1_name:
181                    l_data[l_name] += [x.p1_rank]
182                case x.p2_name:
183                    l_data[l_name] += [x.p2_rank]
184                case x.p3_name:
185                    l_data[l_name] += [x.p3_rank]
186                case x.p4_name:
187                    l_data[l_name] += [x.p4_rank]
188                case _:
189                    l_data[l_name] += [None]
190
191    # 規定打数以下を足切り
192    if g.params.stipulated:
193        for pname in list(l_data.keys()):
194            if sum(x is not None for x in l_data[pname]) < g.params.stipulated:
195                l_data.pop(pname)
196
197    rank_df = pd.DataFrame(l_data.values(), columns=list(df.index), index=list(l_data.keys()))
198
199    # 対象リストが0件になった場合は空のデータフレームを返す
200    if rank_df.empty:
201        return rank_df
202
203    # 対局対戦マトリックス表の作成
204    mtx_df = pd.DataFrame(index=list(l_data.keys()), columns=list(l_data.keys()) + ["total"])
205    sorting_df = pd.DataFrame(index=list(l_data.keys()), columns=["win_per", "count"])
206
207    for idx1 in range(len(rank_df)):
208        p1 = rank_df.iloc[idx1]
209        t_game_count = 0
210        t_win = 0
211        for idx2 in range(len(rank_df)):
212            p2 = rank_df.iloc[idx2]
213            if p1.name == p2.name:
214                mtx_df.loc[f"{p1.name}", f"{p2.name}"] = "---"
215            else:
216                game_count = len(pd.concat([p1, p2], axis=1).dropna())
217                win = (p1 < p2).sum()
218                t_game_count += game_count
219                t_win += win
220
221                if game_count:
222                    winning_per = str(round(float(win / game_count * 100), 1))
223                else:
224                    winning_per = "--.-"
225                mtx_df.loc[f"{p1.name}", f"{p2.name}"] = f"{win}-{game_count - win} ({winning_per}%)"
226
227        if t_game_count:
228            t_winning_per = str(round(float(t_win / t_game_count * 100), 1))
229        else:
230            t_winning_per = "--.-"
231        mtx_df.loc[f"{p1.name}", "total"] = f"{t_win}-{t_game_count - t_win} ({t_winning_per}%)"
232        sorting_df.loc[f"{p1.name}", "win_per"] = t_winning_per
233        sorting_df.loc[f"{p1.name}", "count"] = t_game_count
234
235    # 勝率で並び替え
236    sorting_df["win_per"] = pd.to_numeric(sorting_df["win_per"], errors="coerce")
237    sorting_df["count"] = pd.to_numeric(sorting_df["count"], errors="coerce")
238    sorting_df = sorting_df.sort_values(by=["win_per", "count"], ascending=[False, False])
239    mtx_df = mtx_df.reindex(index=list(sorting_df.index), columns=list(sorting_df.index) + ["total"])
240
241    return mtx_df