libs.data.aggregate

  1"""
  2libs/data/aggregate.py
  3"""
  4
  5import logging
  6from typing import Optional
  7
  8import numpy as np
  9import pandas as pd
 10
 11import libs.global_value as g
 12from libs.data import loader
 13from libs.utils import formatter
 14
 15
 16def game_summary(
 17    filter_items: Optional[list] = None,
 18    drop_items: Optional[list] = None,
 19) -> pd.DataFrame:
 20    """ゲーム結果をサマライズする
 21
 22    Args:
 23        filter_items (Optional[list], optional): 抽出するカラム. Defaults to None.
 24        drop_items (Optional[list], optional): 除外するカラム. Defaults to None.
 25
 26    Returns:
 27        pd.DataFrame: 集計結果
 28    """
 29
 30    # データ収集
 31    df = loader.read_data("SUMMARY_TOTAL")
 32
 33    if isinstance(filter_items, list):
 34        df = df.filter(items=filter_items)
 35
 36    if isinstance(drop_items, list):
 37        df = df.drop(columns=drop_items)
 38
 39    logging.trace(df)  # type: ignore
 40    return df
 41
 42
 43def game_results() -> pd.DataFrame:
 44    """成績を集計する
 45
 46    Returns:
 47        pd.DataFrame: 集計結果
 48    """
 49
 50    # データ収集
 51    df = loader.read_data("SUMMARY_RESULTS")
 52
 53    # Nullが返ってきたときにobject型になるので型変換
 54    df = df.astype({
 55        "東家-平均順位": "float", "南家-平均順位": "float", "西家-平均順位": "float", "北家-平均順位": "float",
 56        "東家-役満和了": "Int64", "南家-役満和了": "Int64", "西家-役満和了": "Int64", "北家-役満和了": "Int64",
 57    }).fillna(0)
 58
 59    # インデックスの振り直し
 60    df = df.reset_index(drop=True)
 61    df.index = df.index + 1
 62
 63    logging.trace(df)  # type: ignore
 64    return df
 65
 66
 67# ランキング
 68def ranking_record() -> pd.DataFrame:
 69    """ランキング集計
 70
 71    Returns:
 72        pd.DataFrame: 集計結果
 73    """
 74
 75    # データ収集
 76    gamedata = loader.read_data("RANKING_RECORD_COUNT")
 77    player_list = gamedata["name"].unique().tolist()
 78
 79    # 連続順位カウント
 80    rank_mask = {
 81        "c_top": {1: 1, 2: 0, 3: 0, 4: 0},  # 連続トップ
 82        "c_top2": {1: 1, 2: 1, 3: 0, 4: 0},  # 連続連対
 83        "c_top3": {1: 1, 2: 1, 3: 1, 4: 0},  # 連続ラス回避
 84        "c_low": {1: 0, 2: 1, 3: 1, 4: 1},  # 連続トップなし
 85        "c_low2": {1: 0, 2: 0, 3: 1, 4: 1},  # 連続逆連対
 86        "c_low4": {1: 0, 2: 0, 3: 0, 4: 1},  # 連続ラス
 87    }
 88
 89    record_df = pd.DataFrame(
 90        {
 91            "name": player_list,
 92            "c_top": [0 for _ in player_list],
 93            "c_top2": [0 for _ in player_list],
 94            "c_top3": [0 for _ in player_list],
 95            "c_low": [0 for _ in player_list],
 96            "c_low2": [0 for _ in player_list],
 97            "c_low4": [0 for _ in player_list],
 98        },
 99        index=player_list
100    )
101
102    for key, val in rank_mask.items():
103        for pname in player_list:
104            tmp_df = pd.DataFrame()
105            tmp_df["flg"] = gamedata.query(
106                "name == @pname"
107            )["順位"].replace(val)
108
109            tmp_df[key] = tmp_df["flg"].groupby(
110                (tmp_df["flg"] != tmp_df["flg"].shift()).cumsum()
111            ).cumcount() + 1
112            tmp_df.loc[tmp_df["flg"] == 0, key] = 0
113            max_key = key.replace("c_", "max_")
114            record_df.at[pname, max_key] = int(tmp_df[[key]].max().values[0])
115
116            # 最終値
117            record_df.at[pname, key] = tmp_df[key].iloc[-1]
118            record_df[max_key] = record_df[max_key].fillna(0).copy().astype("int")
119
120    # 最大値/最小値追加
121    if not gamedata.empty:
122        record_df["point_max"] = gamedata["point_max"].iloc[0]
123        record_df["point_min"] = gamedata["point_min"].iloc[0]
124        record_df["rpoint_max"] = gamedata["rpoint_max"].iloc[0]
125        record_df["rpoint_min"] = gamedata["rpoint_min"].iloc[0]
126
127    logging.trace(record_df)  # type: ignore
128    return record_df
129
130
131def calculation_rating() -> pd.DataFrame:
132    """レーティング集計
133
134    Returns:
135        pd.DataFrame: 集計結果
136    """
137
138    # データ収集
139    df_results = loader.read_data("RANKING_RATINGS").set_index("playtime")
140    df_ratings = pd.DataFrame(index=["initial_rating"] + df_results.index.to_list())  # 記録用
141    last_ratings: dict = {}  # 最終値格納用
142
143    # 獲得スコア
144    score_mapping = {"1": 30.0, "2": 10.0, "3": -10.0, "4": -30.0}
145
146    for x in df_results.itertuples():
147        player_list = (x.p1_name, x.p2_name, x.p3_name, x.p4_name)
148        for player in player_list:
149            if player not in df_ratings.columns:
150                last_ratings[player] = 1500.0
151                df_ratings[player] = np.nan
152                df_ratings.loc["initial_rating", player] = 1500.0
153                df_ratings = df_ratings.copy()
154
155        # 天鳳計算式 (https://tenhou.net/man/#RATING)
156        rank_list = (x.p1_rank, x.p2_rank, x.p3_rank, x.p4_rank,)
157        rating_list = [last_ratings[player] for player in player_list]
158        rating_avg = 1500.0 if np.mean(rating_list) < 1500.0 else np.mean(rating_list)
159
160        for i, player in enumerate(player_list):
161            rating = float(rating_list[i])
162            rank = str(rank_list[i])
163
164            correction_value: float = (rating_avg - rating) / 40
165            if df_ratings[player].count() >= 400:
166                match_correction = 0.2
167            else:
168                match_correction = 1 - df_ratings[player].count() * 0.002
169
170            new_rating = rating + match_correction * (score_mapping[rank] + correction_value)
171
172            last_ratings[player] = new_rating
173            df_ratings.loc[x.Index, player] = new_rating
174
175    # 間引き(集約オプション)
176    if (collection := g.params.get("collection")):
177        ratings = df_ratings[1:]
178        ratings.index = pd.to_datetime(ratings.index)  # DatetimeIndexに変換
179
180        match collection:
181            case "daily":
182                ratings = ratings.resample("D").last().ffill()
183            case "monthly":
184                ratings = ratings.resample("ME").last().ffill()
185            case "yearly":
186                ratings = ratings.resample("YE").last().ffill()
187            case "all":
188                ratings = df_ratings.ffill().tail(1)
189            case _:
190                return df_ratings
191
192        ratings.index = ratings.index.astype(str)
193        df_ratings = pd.concat([df_ratings.head(1), ratings])
194
195    return df_ratings
196
197
198def grade_promotion_check(
199    grade_level: int,
200    point: int,
201    rank: int
202) -> tuple[int, int]:
203    """昇段チェック
204
205    Args:
206        grade_level (int): 現在のレベル(段位)
207        point (int): 現在の昇段ポイント
208        rank (int): 獲得順位
209
210    Returns:
211        tuple[int, int]: チェック後の昇段ポイント, チェック後のレベル(段位)
212    """
213
214    tbl_data = g.cfg.badge.grade.table["table"]
215    new_point = point + int(tbl_data[grade_level]["acquisition"][rank - 1])
216
217    if new_point >= int(tbl_data[grade_level]["point"][1]):  # level up
218        grade_level = min(grade_level + 1, len(tbl_data) - 1)
219        new_point = int(tbl_data[grade_level]["point"][0])  # 初期値
220    elif new_point < 0:  # level down
221        new_point = int(0)
222        if tbl_data[grade_level]["demote"]:
223            grade_level = max(grade_level - 1, 0)
224            new_point = int(tbl_data[grade_level]["point"][0])  # 初期値
225
226    return (new_point, grade_level)
227
228
229# レポート
230def matrix_table() -> pd.DataFrame:
231    """対局対戦マトリックス表の作成
232
233    Returns:
234        pd.DataFrame: 集計結果
235    """
236
237    # データ収集
238    df = loader.read_data("REPORT_MATRIX_TABLE").set_index("playtime")
239
240    # 結果に含まれるプレイヤーのリスト
241    plist = sorted(list(set(
242        df["p1_name"].tolist() + df["p2_name"].tolist() + df["p3_name"].tolist() + df["p4_name"].tolist()
243    )))
244
245    # 順位テーブルの作成
246    l_data: dict = {}
247    for pname in plist:
248        if g.params.get("individual"):  # 個人集計
249            l_name = formatter.name_replace(pname)
250            # プレイヤー指定があるなら対象以外をスキップ
251            if g.params["player_list"]:
252                if l_name not in g.params["player_list"].values():
253                    continue
254            # ゲスト置換
255            if g.params.get("guest_skip"):  # ゲストあり
256                l_name = formatter.name_replace(pname, add_mark=True)
257            else:  # ゲストなし
258                if pname == g.cfg.member.guest_name:
259                    continue
260        else:  # チーム集計
261            l_name = pname
262
263        l_data[l_name] = []
264        for x in df.itertuples():
265            match pname:
266                case x.p1_name:
267                    l_data[l_name] += [x.p1_rank]
268                case x.p2_name:
269                    l_data[l_name] += [x.p2_rank]
270                case x.p3_name:
271                    l_data[l_name] += [x.p3_rank]
272                case x.p4_name:
273                    l_data[l_name] += [x.p4_rank]
274                case _:
275                    l_data[l_name] += [None]
276
277    # 規定打数以下を足切り
278    if g.params["stipulated"]:
279        for pname in list(l_data.keys()):
280            if sum(x is not None for x in l_data[pname]) < g.params["stipulated"]:
281                l_data.pop(pname)
282
283    rank_df = pd.DataFrame(
284        l_data.values(),
285        columns=list(df.index),
286        index=list(l_data.keys())
287    )
288
289    # 対象リストが0件になった場合は空のデータフレームを返す
290    if rank_df.empty:
291        return rank_df
292
293    # 対局対戦マトリックス表の作成
294    mtx_df = pd.DataFrame(
295        index=list(l_data.keys()),
296        columns=list(l_data.keys()) + ["total"]
297    )
298    sorting_df = pd.DataFrame(
299        index=list(l_data.keys()),
300        columns=["win_per", "count"]
301    )
302
303    for idx1 in range(len(rank_df)):
304        p1 = rank_df.iloc[idx1]
305        t_game_count = 0
306        t_win = 0
307        for idx2 in range(len(rank_df)):
308            p2 = rank_df.iloc[idx2]
309            if p1.name == p2.name:
310                mtx_df.loc[f"{p1.name}", f"{p2.name}"] = "---"
311            else:
312                game_count = len(pd.concat([p1, p2], axis=1).dropna())
313                win = (p1 < p2).sum()
314                t_game_count += game_count
315                t_win += win
316
317                if game_count:
318                    winning_per = str(round(float(win / game_count * 100), 1))
319                else:
320                    winning_per = "--.-"
321                mtx_df.loc[f"{p1.name}", f"{p2.name}"] = f"{win}-{game_count - win} ({winning_per}%)"
322
323        if t_game_count:
324            t_winning_per = str(round(float(t_win / t_game_count * 100), 1))
325        else:
326            t_winning_per = "--.-"
327        mtx_df.loc[f"{p1.name}", "total"] = f"{t_win}-{t_game_count - t_win} ({t_winning_per}%)"
328        sorting_df.loc[f"{p1.name}", "win_per"] = t_winning_per
329        sorting_df.loc[f"{p1.name}", "count"] = t_game_count
330
331    # 勝率で並び替え
332    sorting_df["win_per"] = pd.to_numeric(sorting_df["win_per"], errors="coerce")
333    sorting_df["count"] = pd.to_numeric(sorting_df["count"], errors="coerce")
334    sorting_df = sorting_df.sort_values(by=["win_per", "count"], ascending=[False, False])
335    mtx_df = mtx_df.reindex(
336        index=list(sorting_df.index),
337        columns=list(sorting_df.index) + ["total"]
338    )
339
340    return mtx_df