libs.data.aggregate

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