199 lines
9.9 KiB
Python
199 lines
9.9 KiB
Python
import uuid
|
|
|
|
from fastapi import Depends
|
|
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
from common.database.model.models import quotations
|
|
from common.enums import DBWRType, ErrorType, QuotationType, SessionStatus, CardType, CloseReason
|
|
from common.utils.gtime import GTime
|
|
from crud.statistics_crud import StatisticsCRUD, IStatisticsCRUD
|
|
from router.v1.statistics.protocol import (
|
|
Res_StatisticsSummary,
|
|
StatScope,
|
|
StatKpi,
|
|
StatMonthPoint,
|
|
StatMarkupPoint,
|
|
StatOutcome,
|
|
StatParticipation,
|
|
StatTypeRow,
|
|
StatCategory,
|
|
StatCardUsage,
|
|
)
|
|
|
|
WINDOW_MONTHS = 6 # 최근 6개월(당월 포함) 창
|
|
|
|
|
|
class StatisticsService:
|
|
"""통계(성과 분석) 집계. 회사 전체(company)와 내 견적(mine) 두 스코프를 한 응답으로 내린다.
|
|
|
|
전부 파생(저장 안 함) — 조회 때 sessions/items/chats 조인 집계한다. 읽기 전용.
|
|
절감 원천은 '낙찰 세션'(preferred_sp_id) 이며, 여기서 총절감·추이·유형·카테고리·앵커도달률을 모두 파생한다.
|
|
"""
|
|
|
|
def __init__(self, stat_crud: IStatisticsCRUD = Depends(StatisticsCRUD)):
|
|
self.stat_crud = stat_crud
|
|
|
|
async def get_summary(self, company_id: str, user_id: str) -> Res_StatisticsSummary:
|
|
res = Res_StatisticsSummary()
|
|
company_uuid = uuid.UUID(company_id)
|
|
user_uuid = uuid.UUID(user_id)
|
|
labels, window_start = self._window(GTime.UTC())
|
|
|
|
res.company = await self._scope(company_uuid, None, labels, window_start)
|
|
res.mine = await self._scope(company_uuid, user_uuid, labels, window_start)
|
|
return res
|
|
|
|
# ── 스코프 집계 ─────────────────────────────────────────────
|
|
async def _scope(self, company_uuid, owner_uuid, labels, since) -> StatScope:
|
|
scope = StatScope()
|
|
|
|
win_rows = await self._read(lambda s: self.stat_crud.winning_sessions(s, company_uuid, owner_uuid, since))
|
|
outcome_rows = await self._read(lambda s: self.stat_crud.outcome_counts(s, company_uuid, owner_uuid, since))
|
|
type_rows = await self._read(lambda s: self.stat_crud.type_counts(s, company_uuid, owner_uuid, since))
|
|
part_rows = await self._read(lambda s: self.stat_crud.participation_counts(s, company_uuid, owner_uuid, since))
|
|
regen = await self._read_scalar(lambda s: self.stat_crud.regen_avg_round(s, company_uuid, owner_uuid, since))
|
|
markup = await self._read_scalar(lambda s: self.stat_crud.markup_suppression(s, company_uuid, owner_uuid, since))
|
|
markup_rows = await self._read(lambda s: self.stat_crud.markup_suppression_monthly(s, company_uuid, owner_uuid, since))
|
|
card_rows = await self._read(lambda s: self.stat_crud.card_usage(s, company_uuid, owner_uuid, since))
|
|
|
|
scope.trend = self._trend(win_rows, labels)
|
|
scope.markup_trend = [StatMarkupPoint(month=r[0], rate=float(r[1] or 0.0)) for r in markup_rows]
|
|
scope.categories = self._categories(win_rows)
|
|
scope.outcome = self._outcome(outcome_rows)
|
|
scope.participation = self._participation(part_rows)
|
|
scope.type_split = self._type_split(type_rows, win_rows)
|
|
scope.cards = self._cards(card_rows)
|
|
scope.kpi = self._kpi(win_rows, scope.trend, scope.outcome, regen, markup)
|
|
return scope
|
|
|
|
# ── 파생 계산 ───────────────────────────────────────────────
|
|
def _kpi(self, win_rows, trend, outcome, regen, markup) -> StatKpi:
|
|
k = StatKpi()
|
|
total_saving = sum(int(r.target_price) - int(r.bid_price) for r in win_rows)
|
|
total_target = sum(int(r.target_price) for r in win_rows)
|
|
k.total_savings = total_saving
|
|
k.savings_rate = (total_saving / total_target) if total_target else 0.0
|
|
k.anchor_reach_rate = self._anchor_reach(win_rows)
|
|
|
|
closed = outcome.awarded + outcome.open_price + outcome.open_equal + outcome.open_noshow + outcome.open_reject
|
|
k.closed_count = closed
|
|
k.award_rate = (outcome.awarded / closed) if closed else 0.0
|
|
k.regen_avg_round = round(regen, 2)
|
|
k.markup_suppression_rate = round(markup, 4) # 인상억제율(재협상 직전 라운드 투찰가 대비, 파생)
|
|
# 전월 대비: 마지막 두 달 절감액 차(창에 2개월 미만이면 0).
|
|
k.savings_delta_mom = (trend[-1].savings - trend[-2].savings) if len(trend) >= 2 else 0
|
|
return k
|
|
|
|
def _anchor_reach(self, win_rows) -> float:
|
|
# (목표−투찰)/(목표−앵커), 앵커 있고 목표>앵커인 세션만 평균.
|
|
# 세션별로 [0,100%] 클램프 후 평균 — '도달률'이라 앵커 도달=100% 상한(앵커 뚫어도 100%로).
|
|
# (앵커 gap 이 작으면 원비율이 100% 훌쩍 넘어 평균이 왜곡되므로 캡한다.)
|
|
vals = []
|
|
for r in win_rows:
|
|
if r.anchoring_price is None:
|
|
continue
|
|
target, bid, anchor = int(r.target_price), int(r.bid_price), int(r.anchoring_price)
|
|
span = target - anchor
|
|
if span > 0:
|
|
reach = (target - bid) / span
|
|
vals.append(min(1.0, max(0.0, reach)))
|
|
return (sum(vals) / len(vals)) if vals else 0.0
|
|
|
|
def _trend(self, win_rows, labels) -> list:
|
|
bucket = {m: {"savings": 0, "target": 0} for m in labels}
|
|
for r in win_rows:
|
|
m = f"{r.updated_at.year:04d}-{r.updated_at.month:02d}"
|
|
if m in bucket:
|
|
bucket[m]["savings"] += int(r.target_price) - int(r.bid_price)
|
|
bucket[m]["target"] += int(r.target_price)
|
|
out = []
|
|
for m in labels:
|
|
b = bucket[m]
|
|
rate = (b["savings"] / b["target"]) if b["target"] else 0.0
|
|
out.append(StatMonthPoint(month=m, savings=b["savings"], rate=rate))
|
|
return out
|
|
|
|
def _categories(self, win_rows) -> list:
|
|
# TODO: items.category 자유텍스트 그룹 — 표기 흔들리면 지저분. 카테고리 정규화(코드/테이블) 후 개선.
|
|
agg: dict = {}
|
|
for r in win_rows:
|
|
key = r.category or "미분류"
|
|
a = agg.setdefault(key, {"savings": 0, "count": 0})
|
|
a["savings"] += int(r.target_price) - int(r.bid_price)
|
|
a["count"] += 1
|
|
rows = [StatCategory(category=k, savings=v["savings"], count=v["count"]) for k, v in agg.items()]
|
|
rows.sort(key=lambda x: x.savings, reverse=True)
|
|
return rows
|
|
|
|
def _outcome(self, outcome_rows) -> StatOutcome:
|
|
by = {int(cr): int(n) for cr, n in outcome_rows if cr is not None}
|
|
return StatOutcome(
|
|
awarded=by.get(CloseReason.AWARDED.value, 0),
|
|
open_price=by.get(CloseReason.OPEN_PRICE.value, 0),
|
|
open_equal=by.get(CloseReason.OPEN_EQUAL.value, 0),
|
|
open_noshow=by.get(CloseReason.OPEN_NOSHOW.value, 0),
|
|
open_reject=by.get(CloseReason.OPEN_REJECT.value, 0),
|
|
)
|
|
|
|
def _participation(self, part_rows) -> StatParticipation:
|
|
by = {int(st): int(n) for st, n in part_rows if st is not None}
|
|
return StatParticipation(
|
|
bid=by.get(SessionStatus.DONE.value, 0),
|
|
no_participate=by.get(SessionStatus.NOT_PARTICIPATED.value, 0),
|
|
rejected=by.get(SessionStatus.REJECTED.value, 0),
|
|
)
|
|
|
|
def _type_split(self, type_rows, win_rows) -> list:
|
|
# 4개 코드(협상 1·3 / 견적 2·4=1:N)를 2그룹으로 묶는다. 낙찰률·건수=type_counts, 평균절감=낙찰세션.
|
|
grp = {"nego": {"count": 0, "awarded": 0}, "auction": {"count": 0, "awarded": 0}}
|
|
for t, cnt, awarded in type_rows:
|
|
g = "auction" if QuotationType.is_auction(int(t)) else "nego"
|
|
grp[g]["count"] += int(cnt or 0)
|
|
grp[g]["awarded"] += int(awarded or 0)
|
|
|
|
sav = {"nego": [], "auction": []}
|
|
for r in win_rows:
|
|
g = "auction" if QuotationType.is_auction(int(r.type)) else "nego"
|
|
sav[g].append(int(r.target_price) - int(r.bid_price))
|
|
|
|
out = []
|
|
for g, label in (("nego", "협상 (1:1)"), ("auction", "견적 (1:N)")):
|
|
cnt = grp[g]["count"]
|
|
rate = (grp[g]["awarded"] / cnt) if cnt else 0.0
|
|
avg = int(sum(sav[g]) / len(sav[g])) if sav[g] else 0
|
|
out.append(StatTypeRow(label=label, award_rate=rate, avg_savings=avg, count=cnt))
|
|
return out
|
|
|
|
def _cards(self, card_rows) -> list:
|
|
by = {int(ct): int(n) for ct, n in card_rows if ct is not None}
|
|
return [
|
|
StatCardUsage(type="nego", label="협상카드", uses=by.get(CardType.NEGO.value, 0), avg_drop=0),
|
|
StatCardUsage(type="wild", label="와일드카드", uses=by.get(CardType.WILD.value, 0), avg_drop=0),
|
|
]
|
|
|
|
# ── 창(최근 6개월) ─────────────────────────────────────────
|
|
def _window(self, now):
|
|
yy, mm = now.year, now.month
|
|
mm -= (WINDOW_MONTHS - 1)
|
|
while mm <= 0:
|
|
mm += 12
|
|
yy -= 1
|
|
window_start = now.replace(year=yy, month=mm, day=1, hour=0, minute=0, second=0, microsecond=0)
|
|
labels, ly, lm = [], yy, mm
|
|
for _ in range(WINDOW_MONTHS):
|
|
labels.append(f"{ly:04d}-{lm:02d}")
|
|
lm += 1
|
|
if lm > 12:
|
|
lm = 1
|
|
ly += 1
|
|
return labels, window_start
|
|
|
|
# ── DB 실행 헬퍼 ───────────────────────────────────────────
|
|
async def _read(self, fn) -> list:
|
|
err, rows = await DB_SESSION_MNG.execute_lambda(quotations.DBType(), DBWRType.DB_READ.value, fn)
|
|
return rows if err == ErrorType.SUCCESS else []
|
|
|
|
async def _read_scalar(self, fn) -> float:
|
|
err, val = await DB_SESSION_MNG.execute_lambda(quotations.DBType(), DBWRType.DB_READ.value, fn)
|
|
return val if err == ErrorType.SUCCESS else 0.0
|