352 lines
13 KiB
Python
352 lines
13 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", "1")))
|
||
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 normalize_company_key(value: str) -> str:
|
||
import re
|
||
|
||
text = (value or "").upper().replace("Ё", "Е")
|
||
text = re.sub(r"(^|\s)20\d{2}($|\s)", " ", text)
|
||
text = re.sub(r"[^0-9A-ZА-Я]+", " ", text)
|
||
text = re.sub(r"\s+", " ", text)
|
||
return text.strip()
|
||
|
||
|
||
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
|
||
""",
|
||
)
|
||
}
|
||
company_state_map = {
|
||
row["infobase"]: row
|
||
for row in query_rows(
|
||
client,
|
||
"""
|
||
SELECT
|
||
infobase,
|
||
current_status,
|
||
active_locks,
|
||
temp_db_present,
|
||
scheduler_touched,
|
||
current_activity_score
|
||
FROM analytics_1c.v_companies_current
|
||
""",
|
||
)
|
||
}
|
||
excluded_company_keys = {
|
||
str(row["source_company_key"])
|
||
for row in query_rows(
|
||
client,
|
||
"""
|
||
SELECT source_company_key
|
||
FROM analytics_1c.v_company_registry_alias_map
|
||
WHERE exclude_from_portfolio = 1
|
||
""",
|
||
)
|
||
}
|
||
|
||
forecast_rows: list[list[Any]] = []
|
||
signal_rows: list[list[Any]] = []
|
||
|
||
for (infobase, _organization, counterparty), points in grouped.items():
|
||
if normalize_company_key(counterparty) in excluded_company_keys:
|
||
continue
|
||
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)
|
||
company_state = company_state_map.get(infobase, {})
|
||
current_status = str(company_state.get("current_status") or "")
|
||
active_locks = int(company_state.get("active_locks") or 0)
|
||
temp_db_present = int(company_state.get("temp_db_present") or 0)
|
||
scheduler_touched = int(company_state.get("scheduler_touched") or 0)
|
||
current_activity_score = float(company_state.get("current_activity_score") or 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 current_status == "busy" or active_locks > 0 or temp_db_present > 0:
|
||
score = min(85, 45 + active_locks * 5 + temp_db_present * 10)
|
||
signals.append(("base_busy", score, severity_score_to_label(score), f"Файловая база компании {counterparty} занята: status={current_status}, locks={active_locks}, tempDb={temp_db_present}."))
|
||
if scheduler_touched > 0 and current_activity_score >= 15:
|
||
signals.append(("scheduler_activity", 35, "medium", f"По компании {counterparty} есть активность scheduler и повышенный activity score {current_activity_score}."))
|
||
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())
|