302 lines
11 KiB
Python
302 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import math
|
|
import os
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, date, datetime, timedelta
|
|
from statistics import fmean, pstdev
|
|
from typing import Any
|
|
|
|
import clickhouse_connect
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
p = argparse.ArgumentParser(description="Refresh company forecasts and health signals for analytics_1c")
|
|
p.add_argument("--host", default=os.getenv("CLICKHOUSE_HOST", "localhost"))
|
|
p.add_argument("--port", type=int, default=int(os.getenv("CLICKHOUSE_PORT", "8123")))
|
|
p.add_argument("--user", default=os.getenv("CLICKHOUSE_USER", "default"))
|
|
p.add_argument("--password", default=os.getenv("CLICKHOUSE_PASSWORD", ""))
|
|
p.add_argument("--database", default=os.getenv("CLICKHOUSE_DB", "analytics_1c"))
|
|
p.add_argument("--lookback-days", type=int, default=int(os.getenv("AW_1C_COMPANY_LOOKBACK_DAYS", "30")))
|
|
p.add_argument("--min-days", type=int, default=int(os.getenv("AW_1C_COMPANY_MIN_DAYS", "3")))
|
|
p.add_argument("--horizons", default=os.getenv("AW_1C_COMPANY_HORIZONS", "7,30"))
|
|
return p.parse_args()
|
|
|
|
|
|
@dataclass
|
|
class DailyPoint:
|
|
d: date
|
|
docs_total: float
|
|
amount_total: float
|
|
|
|
|
|
def ch_client(args: argparse.Namespace):
|
|
return clickhouse_connect.get_client(
|
|
host=args.host,
|
|
port=args.port,
|
|
username=args.user,
|
|
password=args.password,
|
|
database=args.database,
|
|
)
|
|
|
|
|
|
def query_rows(client, sql: str) -> list[dict[str, Any]]:
|
|
result = client.query(sql)
|
|
return [dict(zip(result.column_names, row)) for row in result.result_rows]
|
|
|
|
|
|
def fill_daily_series(points: list[DailyPoint]) -> list[DailyPoint]:
|
|
if not points:
|
|
return []
|
|
by_day = {p.d: p for p in points}
|
|
current = points[0].d
|
|
end = points[-1].d
|
|
filled: list[DailyPoint] = []
|
|
while current <= end:
|
|
filled.append(by_day.get(current, DailyPoint(current, 0.0, 0.0)))
|
|
current += timedelta(days=1)
|
|
return filled
|
|
|
|
|
|
def linear_slope(values: list[float]) -> float:
|
|
n = len(values)
|
|
if n < 2:
|
|
return 0.0
|
|
x_mean = (n - 1) / 2
|
|
y_mean = fmean(values)
|
|
num = sum((i - x_mean) * (v - y_mean) for i, v in enumerate(values))
|
|
den = sum((i - x_mean) ** 2 for i in range(n))
|
|
if den == 0:
|
|
return 0.0
|
|
return num / den
|
|
|
|
|
|
def build_forecast(values: list[float], horizon: int, min_days: int, lookback_days: int) -> tuple[float, float, float, float, int, str]:
|
|
if len(values) < min_days:
|
|
raise ValueError("not enough data")
|
|
window = values[-min(len(values), lookback_days):]
|
|
baseline = fmean(window)
|
|
slope = linear_slope(window)
|
|
projected = [max(0.0, baseline + slope * step) for step in range(1, horizon + 1)]
|
|
predicted_total = sum(projected)
|
|
predicted_daily = projected[-1] if projected else baseline
|
|
if len(window) > 1 and baseline > 0:
|
|
volatility = pstdev(window) / baseline
|
|
elif len(window) > 1:
|
|
volatility = pstdev(window)
|
|
else:
|
|
volatility = 0.0
|
|
coverage = min(1.0, len(window) / max(lookback_days, 1))
|
|
stability = max(0.15, 1.0 - min(volatility, 1.0))
|
|
confidence = max(0.1, min(0.95, coverage * stability))
|
|
note_parts: list[str] = []
|
|
if len(values) < lookback_days:
|
|
note_parts.append("sparse_history")
|
|
if abs(slope) < 0.01:
|
|
note_parts.append("flat_trend")
|
|
note = ",".join(note_parts) if note_parts else "ok"
|
|
return baseline, slope, predicted_daily, predicted_total, len(window), note
|
|
|
|
|
|
def severity_score_to_label(score: int) -> str:
|
|
if score >= 80:
|
|
return "critical"
|
|
if score >= 60:
|
|
return "high"
|
|
if score >= 35:
|
|
return "medium"
|
|
return "low"
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
client = ch_client(args)
|
|
horizons = [int(x.strip()) for x in args.horizons.split(",") if x.strip()]
|
|
generated_at = datetime.now(UTC).replace(tzinfo=None, microsecond=0)
|
|
|
|
daily_rows = query_rows(
|
|
client,
|
|
"""
|
|
SELECT infobase, organization, counterparty, d, docs_total, amount_total
|
|
FROM analytics_1c.v_counterparty_daily
|
|
ORDER BY infobase, counterparty, d
|
|
""",
|
|
)
|
|
if not daily_rows:
|
|
print("no counterparty rows in analytics_1c.v_counterparty_daily; nothing to refresh")
|
|
return 0
|
|
|
|
grouped: dict[tuple[str, str, str], list[DailyPoint]] = defaultdict(list)
|
|
for row in daily_rows:
|
|
key = (row["infobase"], row["organization"], row["counterparty"])
|
|
grouped[key].append(
|
|
DailyPoint(
|
|
d=row["d"],
|
|
docs_total=float(row["docs_total"] or 0),
|
|
amount_total=float(row["amount_total"] or 0),
|
|
)
|
|
)
|
|
|
|
cases_map = {
|
|
(row["infobase"], row["counterparty"]): int(row["open_cases_total"] or 0)
|
|
for row in query_rows(
|
|
client,
|
|
"""
|
|
SELECT infobase, entity_id AS counterparty, countIf(status != 'closed') AS open_cases_total
|
|
FROM analytics_1c.cases
|
|
WHERE entity_type = 'counterparty'
|
|
GROUP BY infobase, counterparty
|
|
""",
|
|
)
|
|
}
|
|
detections_map = {
|
|
(row["infobase"], row["counterparty"]): int(row["detections_total"] or 0)
|
|
for row in query_rows(
|
|
client,
|
|
"""
|
|
SELECT infobase, entity_id AS counterparty, count() AS detections_total
|
|
FROM analytics_1c.detections
|
|
WHERE entity_type = 'counterparty' AND status != 'closed'
|
|
GROUP BY infobase, counterparty
|
|
""",
|
|
)
|
|
}
|
|
|
|
forecast_rows: list[list[Any]] = []
|
|
signal_rows: list[list[Any]] = []
|
|
|
|
for (infobase, _organization, counterparty), points in grouped.items():
|
|
points.sort(key=lambda p: p.d)
|
|
filled = fill_daily_series(points)
|
|
docs_series = [p.docs_total for p in filled]
|
|
amount_series = [p.amount_total for p in filled]
|
|
if len(filled) < args.min_days:
|
|
continue
|
|
|
|
latest_day = filled[-1].d
|
|
last_7 = filled[-7:]
|
|
prev_7 = filled[-14:-7]
|
|
docs_7d = int(sum(p.docs_total for p in last_7))
|
|
docs_prev_7d = int(sum(p.docs_total for p in prev_7))
|
|
amount_7d = float(sum(p.amount_total for p in last_7))
|
|
amount_prev_7d = float(sum(p.amount_total for p in prev_7))
|
|
days_since_last_activity = (date.today() - latest_day).days
|
|
open_cases_total = cases_map.get((infobase, counterparty), 0)
|
|
detections_total = detections_map.get((infobase, counterparty), 0)
|
|
|
|
for metric, values in (("docs_total", docs_series), ("amount_total", amount_series)):
|
|
for horizon in horizons:
|
|
baseline, slope, predicted_daily, predicted_total, source_days, note = build_forecast(
|
|
values=values,
|
|
horizon=horizon,
|
|
min_days=args.min_days,
|
|
lookback_days=args.lookback_days,
|
|
)
|
|
forecast_rows.append(
|
|
[
|
|
generated_at,
|
|
latest_day,
|
|
infobase,
|
|
counterparty,
|
|
int(horizon),
|
|
metric,
|
|
float(baseline),
|
|
float(slope),
|
|
float(predicted_daily),
|
|
float(predicted_total),
|
|
round(float(max(0.1, min(0.95, 1.0 - abs(slope) / (abs(baseline) + 1.0)))), 4),
|
|
"linear_baseline",
|
|
int(source_days),
|
|
note,
|
|
]
|
|
)
|
|
|
|
signals: list[tuple[str, int, str, str]] = []
|
|
if days_since_last_activity >= 14 and (docs_prev_7d > 0 or amount_prev_7d > 0):
|
|
signals.append(("inactive_company", 85, "high", f"Нет активности по компании {counterparty} уже {days_since_last_activity} дн."))
|
|
if amount_prev_7d > 0 and amount_7d < amount_prev_7d * 0.5:
|
|
signals.append(("amount_drop", 70, "high", f"Объём по компании {counterparty} упал более чем на 50% неделя к неделе."))
|
|
if docs_prev_7d > 0 and docs_7d == 0:
|
|
signals.append(("docs_stopped", 55, "medium", f"По компании {counterparty} прекратился поток документов за последние 7 дней."))
|
|
if open_cases_total > 0:
|
|
signals.append(("open_cases", min(95, 40 + open_cases_total * 10), severity_score_to_label(min(95, 40 + open_cases_total * 10)), f"По компании {counterparty} есть открытые кейсы: {open_cases_total}."))
|
|
if detections_total > 0:
|
|
signals.append(("open_detections", min(90, 35 + detections_total * 5), severity_score_to_label(min(90, 35 + detections_total * 5)), f"По компании {counterparty} есть активные detections: {detections_total}."))
|
|
|
|
for signal_type, score, severity, summary in signals:
|
|
signal_rows.append(
|
|
[
|
|
generated_at,
|
|
infobase,
|
|
counterparty,
|
|
f"{signal_type}:{infobase}:{counterparty}",
|
|
severity,
|
|
int(score),
|
|
signal_type,
|
|
summary,
|
|
float(amount_7d),
|
|
float(amount_prev_7d),
|
|
int(docs_7d),
|
|
int(docs_prev_7d),
|
|
int(max(days_since_last_activity, 0)),
|
|
int(open_cases_total),
|
|
int(detections_total),
|
|
]
|
|
)
|
|
|
|
if forecast_rows:
|
|
client.insert(
|
|
"analytics_1c.company_forecasts",
|
|
forecast_rows,
|
|
column_names=[
|
|
"generated_at",
|
|
"as_of_date",
|
|
"infobase",
|
|
"counterparty",
|
|
"horizon_days",
|
|
"metric",
|
|
"baseline_daily",
|
|
"trend_slope",
|
|
"predicted_daily",
|
|
"predicted_total",
|
|
"confidence",
|
|
"model",
|
|
"source_days",
|
|
"note",
|
|
],
|
|
)
|
|
if signal_rows:
|
|
client.insert(
|
|
"analytics_1c.company_health_signals",
|
|
signal_rows,
|
|
column_names=[
|
|
"generated_at",
|
|
"infobase",
|
|
"counterparty",
|
|
"signal_id",
|
|
"severity",
|
|
"score",
|
|
"signal_type",
|
|
"summary",
|
|
"amount_7d",
|
|
"amount_prev_7d",
|
|
"docs_7d",
|
|
"docs_prev_7d",
|
|
"days_since_last_activity",
|
|
"open_cases_total",
|
|
"detections_total",
|
|
],
|
|
)
|
|
|
|
print(
|
|
f"company intelligence refreshed: forecasts={len(forecast_rows)} signals={len(signal_rows)} generated_at={generated_at.isoformat()}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|