libs.domain.stats

  1"""
  2libs/domain/stats.py
  3"""
  4
  5import textwrap
  6from dataclasses import dataclass, field, fields
  7from typing import TYPE_CHECKING, Any, Literal, Optional, Union, get_type_hints
  8
  9import pandas as pd
 10
 11from libs.utils.timekit import ExtendedDatetime as ExtDt
 12
 13if TYPE_CHECKING:
 14    from libs.domain.placeholder import PlaceholderBuilder
 15
 16
 17@dataclass
 18class StatsDetailed:
 19    """詳細データ"""
 20
 21    mode: Literal[3, 4] = field(default=4)
 22    """集計モード"""
 23    win: int = field(default=0)
 24    """勝ち数"""
 25    lose: int = field(default=0)
 26    """負け数"""
 27    draw: int = field(default=0)
 28    """引き分け数"""
 29    total_point: float = field(default=0.0)
 30    """通算ポイント"""
 31    avg_point: float = field(default=0.0)
 32    """平均ポイント"""
 33
 34    rank1: int = field(default=0)
 35    """1位獲得数"""
 36    rank2: int = field(default=0)
 37    """2位獲得数"""
 38    rank3: int = field(default=0)
 39    """3位獲得数"""
 40    rank4: int = field(default=0)
 41    """4位獲得数"""
 42    flying: int = field(default=0)
 43    """トビ数"""
 44    yakuman: int = field(default=0)
 45    """役満和了数"""
 46
 47    # 最大/最小
 48    rpoint_max: int = field(default=0)
 49    """最大素点"""
 50    rpoint_min: int = field(default=0)
 51    """最小素点"""
 52    point_max: float = field(default=0.0)
 53    """最大獲得ポイント"""
 54    point_min: float = field(default=0.0)
 55    """最小獲得ポイント"""
 56
 57    # 収支データ
 58    score: int = field(default=0)
 59    """最終素点合計"""
 60    score_rank1: int = field(default=0)
 61    """最終素点合計(1位終了時)"""
 62    score_rank2: int = field(default=0)
 63    """最終素点合計(2位終了時)"""
 64    score_rank3: int = field(default=0)
 65    """最終素点合計(3位終了時)"""
 66    score_rank4: int = field(default=0)
 67    """最終素点合計(4位終了時)"""
 68
 69    # レコード
 70    top1_max: int = field(default=0)
 71    """連続1位獲得最大値"""
 72    top1_cur: int = field(default=0)
 73    """連続1位獲得最終値(現在値)"""
 74    top2_max: int = field(default=0)
 75    """連続連対最大値"""
 76    top2_cur: int = field(default=0)
 77    """連続連対最終値(現在値)"""
 78    top3_max: int = field(default=0)
 79    """連続ラス回避獲得最大値"""
 80    top3_cur: int = field(default=0)
 81    """連続ラス回避最終値(現在値)"""
 82    lose2_max: int = field(default=0)
 83    """連続1位なし最大値"""
 84    lose2_cur: int = field(default=0)
 85    """連続1位なし最終値(現在値)"""
 86    lose3_max: int = field(default=0)
 87    """連続逆連対最大値"""
 88    lose3_cur: int = field(default=0)
 89    """連続逆連対最終値(現在値)"""
 90    lose4_max: int = field(default=0)
 91    """連続ラス最大値"""
 92    lose4_cur: int = field(default=0)
 93    """連続ラス最終値(現在値)"""
 94
 95    # 集計範囲
 96    first_game: ExtDt = field(default=ExtDt("1900-01-01 00:00:00"))
 97    """最初の記録時間"""
 98    last_game: ExtDt = field(default=ExtDt("1900-01-01 00:00:00"))
 99    """最後の記録時間"""
100    first_comment: Optional[str] = field(default=None)
101    """最初の記録時間のコメント"""
102    last_comment: Optional[str] = field(default=None)
103    """最後の記録時間のコメント"""
104
105    def avg_balance(self, pattern: str) -> float:
106        """
107        平均収支計算
108
109        Args:
110            pattern (str): 計算パターン
111
112        Returns:
113            float: 計算結果
114
115        """
116        ret: float = 0.0
117
118        match pattern:
119            case "rank1":
120                if self.rank1:
121                    ret = round(self.score_rank1 * 100 / self.rank1, 1)
122            case "rank2":
123                if self.rank2:
124                    ret = round(self.score_rank2 * 100 / self.rank2, 1)
125            case "rank3":
126                if self.rank3:
127                    ret = round(self.score_rank3 * 100 / self.rank3, 1)
128            case "rank4":
129                if self.rank4:
130                    ret = round(self.score_rank4 * 100 / self.rank4, 1)
131            case "top2":
132                if self.rank1 + self.rank2:
133                    ret = round((self.score_rank1 + self.score_rank2) * 100 / (self.rank1 + self.rank2), 1)
134            case "lose2":
135                if self.rank3 + self.rank4:
136                    ret = round((self.score_rank3 + self.score_rank4) * 100 / (self.rank3 + self.rank4), 1)
137            case _:
138                if self.count:
139                    ret = round(self.score * 100 / self.count, 1)
140
141        return ret
142
143    @property
144    def count(self) -> int:
145        """ゲーム数"""
146        match self.mode:
147            case 3:
148                return sum([self.rank1, self.rank2, self.rank3])
149            case 4:
150                return sum([self.rank1, self.rank2, self.rank3, self.rank4])
151
152    @property
153    def rank_avg(self) -> float:
154        """平均順位"""
155        if self.count:
156            match self.mode:
157                case 3:
158                    return round((self.rank1 + self.rank2 * 2 + self.rank3 * 3) / self.count, 2)
159                case 4:
160                    return round((self.rank1 + self.rank2 * 2 + self.rank3 * 3 + self.rank4 * 4) / self.count, 2)
161        return 0.00
162
163    @property
164    def rank_distr(self) -> str:
165        """順位分布(+平均順位)"""
166        match self.mode:
167            case 3:
168                return f"{self.rank1}-{self.rank2}-{self.rank3} ({self.rank_avg:.2f})"
169            case 4:
170                return f"{self.rank1}-{self.rank2}-{self.rank3}-{self.rank4} ({self.rank_avg:.2f})"
171
172    @property
173    def rank_distr2(self) -> str:
174        """順位分布(+ゲーム数)"""
175        match self.mode:
176            case 3:
177                return f"{self.rank1}+{self.rank2}+{self.rank3}={self.count}"
178            case 4:
179                return f"{self.rank1}+{self.rank2}+{self.rank3}+{self.rank4}={self.count}"
180
181    @property
182    def rank1_rate(self) -> float:
183        """1位獲得率"""
184        if self.count:
185            return round(self.rank1 / self.count, 4)
186        return 0.0
187
188    @property
189    def rank2_rate(self) -> float:
190        """2位獲得率"""
191        if self.count:
192            return round(self.rank2 / self.count, 4)
193        return 0.0
194
195    @property
196    def rank3_rate(self) -> float:
197        """3位獲得率"""
198        if self.count:
199            return round(self.rank3 / self.count, 4)
200        return 0.0
201
202    @property
203    def rank4_rate(self) -> float:
204        """4位獲得率"""
205        if self.count:
206            return round(self.rank4 / self.count, 4)
207        return 0.0
208
209    @property
210    def flying_rate(self) -> float:
211        """トビ率"""
212        if self.count:
213            return round(self.flying / self.count, 4)
214        return 0.0
215
216    @property
217    def yakuman_rate(self) -> float:
218        """役満和了率"""
219        if self.count:
220            return round(self.yakuman / self.count, 4)
221        return 0.0
222
223    def update_from_dict(self, data: dict[Any, Any]) -> None:
224        """
225        辞書から値を更新
226
227        Args:
228            data (dict[Any, Any]): 更新データ（キーはフィールド名）
229
230        """
231        type_hints = get_type_hints(self.__class__)
232        for field_obj in fields(self):
233            if field_obj.name not in data:
234                continue
235
236            value = data[field_obj.name]
237            if value is None:
238                setattr(self, field_obj.name, None)
239                continue
240
241            field_type = type_hints.get(field_obj.name, type(None))
242
243            # 型別の変換処理
244            if field_type in (int, float, str, bool):
245                setattr(self, field_obj.name, field_type(value))
246            elif field_type is ExtDt or (hasattr(field_type, "__origin__") and field_type.__origin__ is Union):
247                # ExtDtまたはOptional型
248                if hasattr(field_type, "__args__"):
249                    # Optional[X]の場合、Xの型を取得
250                    inner_type = next(arg for arg in field_type.__args__ if arg is not type(None))
251                    setattr(self, field_obj.name, inner_type(value) if value is not None else None)
252                else:
253                    setattr(self, field_obj.name, ExtDt(value))
254            else:
255                setattr(self, field_obj.name, value)
256
257    def war_record(self) -> str:
258        """戦績結果"""
259        return f"{self.count} 戦 ({self.win} 勝 {self.lose} 敗 {self.draw} 分)"
260
261    def best_record(self) -> str:
262        """ベストレコード"""
263        rpoint_max = f"{self.rpoint_max * 100:+}点".replace("-", "▲") if self.rpoint_max else "記録なし"
264        point_max = f"{self.point_max:+.1f}pt".replace("-", "▲") if self.point_max else "記録なし"
265
266        ret = textwrap.dedent(f"""\
267            連続トップ：{self._work(self.top1_cur, self.top1_max)}
268            連続連対：{self._work(self.top2_cur, self.top2_max)}
269            連続ラス回避：{self._work(self.top3_cur, self.top3_max)}
270            最大素点：{rpoint_max}
271            最大獲得ポイント：{point_max}
272            """)
273        return ret.strip()
274
275    def worst_record(self) -> str:
276        """ワーストレコード"""
277        rpoint_min = f"{self.rpoint_min * 100:+}点".replace("-", "▲") if self.rpoint_min else "記録なし"
278        point_min = f"{self.point_min:+.1f}pt".replace("-", "▲") if self.point_min else "記録なし"
279
280        ret = textwrap.dedent(f"""\
281            連続ラス：{self._work(self.lose4_cur, self.lose4_max)}
282            連続逆連対：{self._work(self.lose3_cur, self.lose3_max)}
283            連続トップなし：{self._work(self.lose2_cur, self.lose2_max)}
284            最大素点：{rpoint_min}
285            最大獲得ポイント：{point_min}
286            """)
287        return ret.strip()
288
289    def _work(self, c_num: int, m_num: int) -> str:
290        """
291        単位設定
292
293        Args:
294            c_num (int): 現在値
295            m_num (int): 最大値
296
297        Returns:
298            str: 生成文字列
299
300        """
301        c_str = f"{c_num} 回目" if c_num else f"{c_num} 回"
302        if m_num:
303            m_str = "最大 1 回" if m_num == 1 else f"最大 {m_num} 連続"
304            if m_num == c_num:
305                m_str = "記録更新中"
306        else:
307            m_str = "記録なし"
308
309        return f"{c_str} ({m_str})"
310
311
312@dataclass
313class StatsInfo:
314    """成績情報"""
315
316    name: str = field(default="")
317    """プレイヤー名/チーム名"""
318
319    # 統計データ
320    seat0: StatsDetailed = field(default_factory=StatsDetailed)
321    """全席"""
322    seat1: StatsDetailed = field(default_factory=StatsDetailed)
323    """東家"""
324    seat2: StatsDetailed = field(default_factory=StatsDetailed)
325    """南家"""
326    seat3: StatsDetailed = field(default_factory=StatsDetailed)
327    """西家"""
328    seat4: StatsDetailed = field(default_factory=StatsDetailed)
329    """北家"""
330
331    # 検索条件
332    mode: Literal[3, 4] = field(default=4)
333    """集計モード"""
334    rule_version: list[str] = field(default_factory=list)
335    """ルールセット"""
336    # 検索範囲
337    starttime: ExtDt = field(default=ExtDt("1900-01-01 00:00:00"))
338    endtime: ExtDt = field(default=ExtDt("1900-01-01 00:00:00"))
339
340    # 検索ワード
341    search_word: str = field(default="")
342
343    # 取り込みデータ
344    result_df: pd.DataFrame = field(default_factory=pd.DataFrame)
345    record_df: pd.DataFrame = field(default_factory=pd.DataFrame)
346
347    def read(self, params: "PlaceholderBuilder") -> None:
348        """
349        データ読み込み
350
351        Args:
352            params (PlaceholderBuilder): プレースホルダ
353
354        """
355        self.result_df = params.read_data("RESULTS_INFO")
356        self.record_df = params.read_data("RECORD_INFO")
357
358        if self.result_df.empty or self.record_df.empty:
359            return
360
361        self.set_parameter(**params.placeholder())
362        self.set_data(self.result_df)
363        self.set_data(self.record_df)
364
365    def set_data(self, df: "pd.DataFrame") -> None:
366        """
367        集計結果取り込み
368
369        Args:
370            df (pd.DataFrame): 集計結果
371
372        """
373        seat_map = {0: self.seat0, 1: self.seat1, 2: self.seat2, 3: self.seat3, 4: self.seat4}
374
375        for _, row in df.iterrows():
376            if "id" in df.columns:
377                seat_id = row["id"]
378                if isinstance(seat_id, int) and seat_id in seat_map:
379                    seat_map[seat_id].update_from_dict(row.to_dict())
380
381    def set_parameter(self, **kwargs: Any) -> None:
382        """パラメータ取り込み"""
383        if "mode" in kwargs and isinstance(kwargs["mode"], int):
384            if kwargs["mode"] in (3, 4):
385                self.mode = kwargs["mode"]  # type: ignore[assignment]
386                self.seat0.mode = self.mode
387                self.seat1.mode = self.mode
388                self.seat2.mode = self.mode
389                self.seat3.mode = self.mode
390                self.seat4.mode = self.mode
391            else:
392                raise ValueError(f"Unsupported mode: {kwargs['mode']}")
393
394        if "rule_list" in kwargs and isinstance(kwargs["rule_list"], list):
395            self.rule_version = kwargs["rule_list"]
396        if "rule_set" in kwargs and isinstance(kwargs["rule_set"], dict):
397            self.rule_version = list(kwargs["rule_set"].values())
398        if "player_name" in kwargs and isinstance(kwargs["player_name"], str):
399            self.name = kwargs["player_name"]
400        if "starttime" in kwargs and isinstance(kwargs["starttime"], (ExtDt, str)):
401            self.starttime = ExtDt(kwargs["starttime"])
402        if "endtime" in kwargs and isinstance(kwargs["endtime"], (ExtDt, str)):
403            self.endtime = ExtDt(kwargs["endtime"])
404        if "search_word" in kwargs and isinstance(kwargs["search_word"], str):
405            self.search_word = kwargs["search_word"]
406
407    @property
408    def rank_distr_list(self) -> list[str]:
409        """座席別順位分布(平均順位)"""
410        return [
411            self.seat1.rank_distr,
412            self.seat2.rank_distr,
413            self.seat3.rank_distr,
414            self.seat4.rank_distr,
415        ][: self.mode]
416
417    @property
418    def rank_distr_list2(self) -> list[str]:
419        """座席別順位分布(ゲーム数)"""
420        return [
421            self.seat1.rank_distr2,
422            self.seat2.rank_distr2,
423            self.seat3.rank_distr2,
424            self.seat4.rank_distr2,
425        ][: self.mode]
426
427    @property
428    def rank_avg_list(self) -> list[float]:
429        """座席別平均順位"""
430        return [
431            self.seat1.rank_avg,
432            self.seat2.rank_avg,
433            self.seat3.rank_avg,
434            self.seat4.rank_avg,
435        ][: self.mode]
436
437    @property
438    def flying_list(self) -> list[int]:
439        """座席別トビ率"""
440        return [
441            self.seat1.flying,
442            self.seat2.flying,
443            self.seat3.flying,
444            self.seat4.flying,
445        ][: self.mode]
446
447    @property
448    def yakuman_list(self) -> list[int]:
449        """座席別役満和了率"""
450        return [
451            self.seat1.yakuman,
452            self.seat2.yakuman,
453            self.seat3.yakuman,
454            self.seat4.yakuman,
455        ][: self.mode]
456
457    @property
458    def summary(self) -> pd.DataFrame:
459        """成績サマリ"""
460        ret_df = pd.DataFrame(
461            {
462                "count": [self.seat0.count],
463                "war_record": [f"{self.seat0.win}-{self.seat0.lose}-{self.seat0.draw}"],
464                "rank_avg": [self.seat0.rank_avg],
465                "total_point": [f"{self.seat0.total_point:+.1f}pt".replace("-", "▲")],
466                "avg_point": [f"{self.seat0.avg_point:+.1f}pt".replace("-", "▲")],
467                "top2_rate-count": [f"{(self.seat0.rank1 + self.seat0.rank2) / self.seat0.count:.2%}({self.seat0.rank1 + self.seat0.rank2})"],
468                "top3_rate-count": [
469                    f"{(self.seat0.rank1 + self.seat0.rank2 + self.seat0.rank3) / self.seat0.count:.2%}"
470                    + f"({self.seat0.rank1 + self.seat0.rank2 + self.seat0.rank3})",
471                ],
472                "rank1_rate-count": [f"{self.seat0.rank1_rate:.2%}({self.seat0.rank1})"],
473                "rank2_rate-count": [f"{self.seat0.rank2_rate:.2%}({self.seat0.rank2})"],
474                "rank3_rate-count": [f"{self.seat0.rank3_rate:.2%}({self.seat0.rank3})"],
475                "rank4_rate-count": [f"{self.seat0.rank4_rate:.2%}({self.seat0.rank4})"],
476                "flying_rate-count": [f"{self.seat0.flying_rate:.2%}({self.seat0.flying})"],
477                "yakuman_rate-count": [f"{self.seat0.yakuman_rate:.2%}({self.seat0.yakuman})"],
478                "avg_balance": [f"{self.seat0.avg_balance('all'):+.1f}点".replace("-", "▲")],
479                "top2_balance": [f"{self.seat0.avg_balance('top2'):+.1f}点".replace("-", "▲")],
480                "lose2_balance": [f"{self.seat0.avg_balance('lose2'):+.1f}点".replace("-", "▲")],
481                "rank1_balance": [f"{self.seat0.avg_balance('rank1'):+.1f}点".replace("-", "▲")],
482                "rank2_balance": [f"{self.seat0.avg_balance('rank2'):+.1f}点".replace("-", "▲")],
483                "rank3_balance": [f"{self.seat0.avg_balance('rank3'):+.1f}点".replace("-", "▲")],
484                "rank4_balance": [f"{self.seat0.avg_balance('rank4'):+.1f}点".replace("-", "▲")],
485                "top1_max": [f"{self.seat0.top1_max}連続"],
486                "top2_max": [f"{self.seat0.top2_max}連続"],
487                "top3_max": [f"{self.seat0.top3_max}連続"],
488                "lose2_max": [f"{self.seat0.lose2_max}連続"],
489                "lose3_max": [f"{self.seat0.lose3_max}連続"],
490                "lose4_max": [f"{self.seat0.lose4_max}連続"],
491                "rpoint_max": [f"{self.seat0.rpoint_max * 100}点".replace("-", "▲")],
492                "point_max": [f"{self.seat0.point_max:+.1f}pt".replace("-", "▲")],
493                "rpoint_min": [f"{self.seat0.rpoint_min * 100}点".replace("-", "▲")],
494                "point_min": [f"{self.seat0.point_min:+.1f}pt".replace("-", "▲")],
495            },
496            index=[self.name],
497        )
498
499        return ret_df