external_demand/app/hot_content/wxindex_trend.py

"""微信指数趋势计算工具。"""

from __future__ import annotations

import math
from typing import Any


MIN_TREND_POINTS = 4
MAX_TREND_POINTS = 7
UP_FIT_CHANGE_RATE = 0.04
UP_WINDOW_CHANGE_RATE = 0.02
DOWN_FIT_CHANGE_RATE = -0.04
DOWN_WINDOW_CHANGE_RATE = -0.02


def _median(values: list[float]) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    mid = len(ordered) // 2
    if len(ordered) % 2:
        return ordered[mid]
    return (ordered[mid - 1] + ordered[mid]) / 2


def _extract_recent_scores(series: list[dict[str, Any]]) -> list[float]:
    scored_rows: list[tuple[str, int, float]] = []
    for index, row in enumerate(series):
        if not isinstance(row, dict):
            continue
        try:
            score = float(row.get("total_score"))
        except (TypeError, ValueError):
            continue
        if math.isnan(score) or score < 0:
            continue
        ymd = str(row.get("ymd") or "").strip()
        scored_rows.append((ymd, index, score))

    scored_rows.sort(key=lambda item: (item[0], item[1]))
    return [score for _, _, score in scored_rows[-MAX_TREND_POINTS:]]


def _theil_sen_slope(values: list[float]) -> float:
    slopes: list[float] = []
    for start_index, start_value in enumerate(values):
        for end_index in range(start_index + 1, len(values)):
            slopes.append((values[end_index] - start_value) / (end_index - start_index))
    return _median(slopes)


def calc_wxindex_trend(series: list[dict[str, Any]]) -> str:
    """按最近 7 天整体走势计算趋势，避免被最后一天波动误导。"""
    scores = _extract_recent_scores(series)
    if len(scores) < MIN_TREND_POINTS:
        return "未知"

    log_scores = [math.log1p(score) for score in scores]
    slope = _theil_sen_slope(log_scores)
    fit_change_rate = math.expm1(slope * (len(log_scores) - 1))

    early_avg = sum(scores[:3]) / 3
    late_avg = sum(scores[-3:]) / 3
    window_change_rate = (late_avg - early_avg) / max(early_avg, 1.0)

    if fit_change_rate >= UP_FIT_CHANGE_RATE and window_change_rate >= UP_WINDOW_CHANGE_RATE:
        return "上升"
    if (
        fit_change_rate <= DOWN_FIT_CHANGE_RATE
        and window_change_rate <= DOWN_WINDOW_CHANGE_RATE
    ):
        return "下降"
    return "平稳"