|
@@ -4,15 +4,17 @@
|
|
|
流程:
|
|
流程:
|
|
|
1. 比对当天 ODPS / MySQL 行数,相同则跳过写入
|
|
1. 比对当天 ODPS / MySQL 行数,相同则跳过写入
|
|
|
2. 有差异时拉取 ODPS,按 (strategy, demand_id) 只插入缺失、删除多余
|
|
2. 有差异时拉取 ODPS,按 (strategy, demand_id) 只插入缺失、删除多余
|
|
|
-3. 查询当天全部 demand_name,按空格分词写入 set
|
|
|
|
|
-4. 过滤 demand_belong_category 中已存在的词
|
|
|
|
|
-5. 剩余词按 100 词一批调用 demand_belong_category_agent
|
|
|
|
|
-6. 遍历 demand_belong_category 全部词,按策略统计热度写入 demand_popularity_stats
|
|
|
|
|
|
|
+3. 拉取近 7 日真实 ROV/VOV,按特征值匹配回填 real_rov_7d / real_vov_7d
|
|
|
|
|
+4. 查询当天全部 demand_name,按空格分词写入 set
|
|
|
|
|
+5. 过滤 demand_belong_category 中已存在的词
|
|
|
|
|
+6. 剩余词按 100 词一批调用 demand_belong_category_agent
|
|
|
|
|
+7. 遍历 demand_belong_category 全部词,按策略与真实 ROV/VOV 写入 demand_popularity_stats(avg/count)
|
|
|
|
|
+8. 基于三表计算整棵类目树节点加权平均分,写入 category_tree_weight
|
|
|
"""
|
|
"""
|
|
|
from __future__ import annotations
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
import logging
|
|
import logging
|
|
|
-from datetime import datetime
|
|
|
|
|
|
|
+from datetime import datetime, timedelta
|
|
|
from decimal import Decimal
|
|
from decimal import Decimal
|
|
|
from typing import Any
|
|
from typing import Any
|
|
|
|
|
|
|
@@ -26,11 +28,17 @@ from supply_infra.db.repositories.demand_popularity_stats_repo import (
|
|
|
from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
|
|
from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
|
|
|
from supply_infra.db.session import get_session
|
|
from supply_infra.db.session import get_session
|
|
|
from supply_infra.odps.client import get_odps_client
|
|
from supply_infra.odps.client import get_odps_client
|
|
|
|
|
+from supply_infra.scheduler.jobs.compute_category_tree_weight import (
|
|
|
|
|
+ compute_category_tree_weight,
|
|
|
|
|
+)
|
|
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
_WORD_BATCH_SIZE = 100
|
|
_WORD_BATCH_SIZE = 100
|
|
|
_STATS_UPSERT_BATCH = 200
|
|
_STATS_UPSERT_BATCH = 200
|
|
|
|
|
+_REAL_METRIC_LIMIT = 1000
|
|
|
|
|
+_REAL_METRIC_LOOKBACK_DAYS = 7
|
|
|
|
|
+_GLOBAL_FEATURE_VALUE = "全局SUM"
|
|
|
|
|
|
|
|
# 策略名 → 统计字段前缀;去年同期阳历/阴历合并为 plat_ly_pop
|
|
# 策略名 → 统计字段前缀;去年同期阳历/阴历合并为 plat_ly_pop
|
|
|
_STRATEGY_METRIC: dict[str, str] = {
|
|
_STRATEGY_METRIC: dict[str, str] = {
|
|
@@ -42,6 +50,16 @@ _STRATEGY_METRIC: dict[str, str] = {
|
|
|
}
|
|
}
|
|
|
_TARGET_STRATEGIES = list(_STRATEGY_METRIC.keys())
|
|
_TARGET_STRATEGIES = list(_STRATEGY_METRIC.keys())
|
|
|
_METRIC_KEYS = ("ext_pop", "plat_sust_pop", "plat_ly_pop", "recent_pop")
|
|
_METRIC_KEYS = ("ext_pop", "plat_sust_pop", "plat_ly_pop", "recent_pop")
|
|
|
|
|
+_REAL_METRIC_KEYS = ("real_rov_7d", "real_vov_7d")
|
|
|
|
|
+_ALL_METRIC_KEYS = _METRIC_KEYS + _REAL_METRIC_KEYS
|
|
|
|
|
+_METRIC_DECIMAL_PLACES: dict[str, int] = {
|
|
|
|
|
+ "ext_pop": 2,
|
|
|
|
|
+ "plat_sust_pop": 2,
|
|
|
|
|
+ "plat_ly_pop": 2,
|
|
|
|
|
+ "recent_pop": 2,
|
|
|
|
|
+ "real_rov_7d": 4,
|
|
|
|
|
+ "real_vov_7d": 4,
|
|
|
|
|
+}
|
|
|
|
|
|
|
|
RowKey = tuple[str, str]
|
|
RowKey = tuple[str, str]
|
|
|
|
|
|
|
@@ -160,30 +178,128 @@ def _sync_diff(partition_date: str) -> dict[str, Any]:
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
-def _calc_metric_stats(weights: list[float]) -> tuple[Decimal, Decimal, int]:
|
|
|
|
|
|
|
+def _to_float(value: Any) -> float | None:
|
|
|
|
|
+ if value is None:
|
|
|
|
|
+ return None
|
|
|
|
|
+ try:
|
|
|
|
|
+ return float(value)
|
|
|
|
|
+ except (TypeError, ValueError):
|
|
|
|
|
+ return None
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _pick_better_real_metric(
|
|
|
|
|
+ current: dict[str, Any] | None,
|
|
|
|
|
+ candidate: dict[str, Any],
|
|
|
|
|
+) -> dict[str, Any]:
|
|
|
|
|
+ """相同特征值时保留 rov_diff、vov_diff 更高的一条。"""
|
|
|
|
|
+ if current is None:
|
|
|
|
|
+ return candidate
|
|
|
|
|
+
|
|
|
|
|
+ current_key = (
|
|
|
|
|
+ _to_float(current.get("rov_diff")) or 0.0,
|
|
|
|
|
+ _to_float(current.get("vov_diff")) or 0.0,
|
|
|
|
|
+ )
|
|
|
|
|
+ candidate_key = (
|
|
|
|
|
+ _to_float(candidate.get("rov_diff")) or 0.0,
|
|
|
|
|
+ _to_float(candidate.get("vov_diff")) or 0.0,
|
|
|
|
|
+ )
|
|
|
|
|
+ return candidate if candidate_key > current_key else current
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _build_real_metrics_by_feature(
|
|
|
|
|
+ raw_rows: list[dict[str, Any]],
|
|
|
|
|
+) -> dict[str, tuple[float | None, float | None]]:
|
|
|
|
|
+ """按特征值去重,映射为 demand_name → (real_rov_7d, real_vov_7d)。"""
|
|
|
|
|
+ best_by_feature: dict[str, dict[str, Any]] = {}
|
|
|
|
|
+ for row in raw_rows:
|
|
|
|
|
+ feature = row.get("特征值")
|
|
|
|
|
+ if feature is None:
|
|
|
|
|
+ continue
|
|
|
|
|
+ feature_name = str(feature).strip()
|
|
|
|
|
+ if not feature_name or feature_name == _GLOBAL_FEATURE_VALUE:
|
|
|
|
|
+ continue
|
|
|
|
|
+ best_by_feature[feature_name] = _pick_better_real_metric(
|
|
|
|
|
+ best_by_feature.get(feature_name),
|
|
|
|
|
+ row,
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ return {
|
|
|
|
|
+ feature: (_to_float(row.get("rov")), _to_float(row.get("vov")))
|
|
|
|
|
+ for feature, row in best_by_feature.items()
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def enrich_real_rov_vov_7d(biz_dt: str) -> dict[str, Any]:
|
|
|
"""
|
|
"""
|
|
|
- 计算单策略 max / avg / count。
|
|
|
|
|
- 权重为 0 不参与平均值;全部为 0(或无有效权重)时 avg 默认为 0。
|
|
|
|
|
|
|
+ 从 ODPS 拉取执行日及往前 7 日的真实 ROV/VOV,按特征值匹配回填 MySQL。
|
|
|
|
|
+
|
|
|
|
|
+ 相同特征值保留 rov_diff、vov_diff 更高的记录。
|
|
|
"""
|
|
"""
|
|
|
- count = len(weights)
|
|
|
|
|
- if count == 0:
|
|
|
|
|
- return Decimal("0.00"), Decimal("0.00"), 0
|
|
|
|
|
|
|
+ dt_right = biz_dt
|
|
|
|
|
+ dt_left = (
|
|
|
|
|
+ datetime.strptime(biz_dt, "%Y%m%d") - timedelta(days=_REAL_METRIC_LOOKBACK_DAYS)
|
|
|
|
|
+ ).strftime("%Y%m%d")
|
|
|
|
|
|
|
|
- max_val = max(weights)
|
|
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ "Enrich real rov/vov: biz_dt=%s range=%s~%s limit=%d",
|
|
|
|
|
+ biz_dt,
|
|
|
|
|
+ dt_left,
|
|
|
|
|
+ dt_right,
|
|
|
|
|
+ _REAL_METRIC_LIMIT,
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ odps = get_odps_client()
|
|
|
|
|
+ raw_rows = odps.fetch_real_rov_vov_7d(
|
|
|
|
|
+ dt_left=dt_left,
|
|
|
|
|
+ dt_right=dt_right,
|
|
|
|
|
+ limit=_REAL_METRIC_LIMIT,
|
|
|
|
|
+ )
|
|
|
|
|
+ metrics_by_name = _build_real_metrics_by_feature(raw_rows)
|
|
|
|
|
+
|
|
|
|
|
+ with get_session() as session:
|
|
|
|
|
+ updated = MultiDemandPoolDiRepository(session).update_real_metrics_by_demand_name(
|
|
|
|
|
+ biz_dt,
|
|
|
|
|
+ metrics_by_name,
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ result = {
|
|
|
|
|
+ "dt_left": dt_left,
|
|
|
|
|
+ "dt_right": dt_right,
|
|
|
|
|
+ "odps_rows": len(raw_rows),
|
|
|
|
|
+ "unique_features": len(metrics_by_name),
|
|
|
|
|
+ "updated_rows": updated,
|
|
|
|
|
+ }
|
|
|
|
|
+ logger.info("Enrich real rov/vov completed: %s", result)
|
|
|
|
|
+ return result
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _calc_metric_stats(
|
|
|
|
|
+ weights: list[float],
|
|
|
|
|
+ places: int = 2,
|
|
|
|
|
+) -> tuple[Decimal, int]:
|
|
|
|
|
+ """
|
|
|
|
|
+ 计算单策略 avg / count。
|
|
|
|
|
+ 权重为 0 不参与平均值;全部为 0(或无有效权重)时 avg=0、count=0。
|
|
|
|
|
+ """
|
|
|
nonzero = [w for w in weights if w != 0]
|
|
nonzero = [w for w in weights if w != 0]
|
|
|
- avg_val = sum(nonzero) / len(nonzero) if nonzero else 0.0
|
|
|
|
|
|
|
+ if not nonzero:
|
|
|
|
|
+ zero = Decimal("0").quantize(Decimal("0." + "0" * places))
|
|
|
|
|
+ return zero, 0
|
|
|
|
|
+
|
|
|
|
|
+ avg_val = sum(nonzero) / len(nonzero)
|
|
|
|
|
+ q = Decimal("0." + "0" * places)
|
|
|
return (
|
|
return (
|
|
|
- Decimal(str(round(max_val, 2))),
|
|
|
|
|
- Decimal(str(round(avg_val, 2))),
|
|
|
|
|
- count,
|
|
|
|
|
|
|
+ Decimal(str(round(avg_val, places))).quantize(q),
|
|
|
|
|
+ len(nonzero),
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
def _empty_metric_stats() -> dict[str, Decimal | int]:
|
|
def _empty_metric_stats() -> dict[str, Decimal | int]:
|
|
|
result: dict[str, Decimal | int] = {}
|
|
result: dict[str, Decimal | int] = {}
|
|
|
- for key in _METRIC_KEYS:
|
|
|
|
|
- result[f"{key}_max"] = Decimal("0.00")
|
|
|
|
|
- result[f"{key}_avg"] = Decimal("0.00")
|
|
|
|
|
|
|
+ for key in _ALL_METRIC_KEYS:
|
|
|
|
|
+ places = _METRIC_DECIMAL_PLACES[key]
|
|
|
|
|
+ zero = Decimal("0").quantize(Decimal("0." + "0" * places))
|
|
|
|
|
+ result[f"{key}_avg"] = zero
|
|
|
result[f"{key}_count"] = 0
|
|
result[f"{key}_count"] = 0
|
|
|
return result
|
|
return result
|
|
|
|
|
|
|
@@ -193,8 +309,9 @@ def _build_stats_row(
|
|
|
demand_word_name: str,
|
|
demand_word_name: str,
|
|
|
biz_dt: str,
|
|
biz_dt: str,
|
|
|
strategy_weights: list[tuple[str, float | None]],
|
|
strategy_weights: list[tuple[str, float | None]],
|
|
|
|
|
+ real_metrics: list[tuple[float | None, float | None]],
|
|
|
) -> dict[str, Any]:
|
|
) -> dict[str, Any]:
|
|
|
- """按策略分组后汇总为 demand_popularity_stats 一行。"""
|
|
|
|
|
|
|
+ """按策略分组后汇总为 demand_popularity_stats 一行(含真实 ROV/VOV)。"""
|
|
|
grouped: dict[str, list[float]] = {key: [] for key in _METRIC_KEYS}
|
|
grouped: dict[str, list[float]] = {key: [] for key in _METRIC_KEYS}
|
|
|
for strategy, weight in strategy_weights:
|
|
for strategy, weight in strategy_weights:
|
|
|
metric = _STRATEGY_METRIC.get(strategy)
|
|
metric = _STRATEGY_METRIC.get(strategy)
|
|
@@ -202,6 +319,14 @@ def _build_stats_row(
|
|
|
continue
|
|
continue
|
|
|
grouped[metric].append(float(weight))
|
|
grouped[metric].append(float(weight))
|
|
|
|
|
|
|
|
|
|
+ rov_values: list[float] = []
|
|
|
|
|
+ vov_values: list[float] = []
|
|
|
|
|
+ for rov, vov in real_metrics:
|
|
|
|
|
+ if rov is not None:
|
|
|
|
|
+ rov_values.append(float(rov))
|
|
|
|
|
+ if vov is not None:
|
|
|
|
|
+ vov_values.append(float(vov))
|
|
|
|
|
+
|
|
|
row: dict[str, Any] = {
|
|
row: dict[str, Any] = {
|
|
|
"demand_category_id": demand_category_id,
|
|
"demand_category_id": demand_category_id,
|
|
|
"demand_word_name": demand_word_name[:128],
|
|
"demand_word_name": demand_word_name[:128],
|
|
@@ -209,17 +334,29 @@ def _build_stats_row(
|
|
|
**_empty_metric_stats(),
|
|
**_empty_metric_stats(),
|
|
|
}
|
|
}
|
|
|
for key in _METRIC_KEYS:
|
|
for key in _METRIC_KEYS:
|
|
|
- max_val, avg_val, count = _calc_metric_stats(grouped[key])
|
|
|
|
|
- row[f"{key}_max"] = max_val
|
|
|
|
|
|
|
+ avg_val, count = _calc_metric_stats(
|
|
|
|
|
+ grouped[key], places=_METRIC_DECIMAL_PLACES[key]
|
|
|
|
|
+ )
|
|
|
row[f"{key}_avg"] = avg_val
|
|
row[f"{key}_avg"] = avg_val
|
|
|
row[f"{key}_count"] = count
|
|
row[f"{key}_count"] = count
|
|
|
|
|
+
|
|
|
|
|
+ rov_avg, rov_count = _calc_metric_stats(
|
|
|
|
|
+ rov_values, places=_METRIC_DECIMAL_PLACES["real_rov_7d"]
|
|
|
|
|
+ )
|
|
|
|
|
+ vov_avg, vov_count = _calc_metric_stats(
|
|
|
|
|
+ vov_values, places=_METRIC_DECIMAL_PLACES["real_vov_7d"]
|
|
|
|
|
+ )
|
|
|
|
|
+ row["real_rov_7d_avg"] = rov_avg
|
|
|
|
|
+ row["real_rov_7d_count"] = rov_count
|
|
|
|
|
+ row["real_vov_7d_avg"] = vov_avg
|
|
|
|
|
+ row["real_vov_7d_count"] = vov_count
|
|
|
return row
|
|
return row
|
|
|
|
|
|
|
|
|
|
|
|
|
def compute_popularity_stats(biz_dt: str) -> dict[str, Any]:
|
|
def compute_popularity_stats(biz_dt: str) -> dict[str, Any]:
|
|
|
"""
|
|
"""
|
|
|
遍历 demand_belong_category 全部词,各自 LIKE 查询当天指定策略权重,
|
|
遍历 demand_belong_category 全部词,各自 LIKE 查询当天指定策略权重,
|
|
|
- 分策略统计后写入 demand_popularity_stats。
|
|
|
|
|
|
|
+ 并汇总真实 ROV/VOV,写入 demand_popularity_stats(avg/count)。
|
|
|
|
|
|
|
|
Args:
|
|
Args:
|
|
|
biz_dt: 业务日期 (YYYYMMDD)
|
|
biz_dt: 业务日期 (YYYYMMDD)
|
|
@@ -244,7 +381,10 @@ def compute_popularity_stats(biz_dt: str) -> dict[str, Any]:
|
|
|
matches = pool_repo.list_weights_by_name_like(
|
|
matches = pool_repo.list_weights_by_name_like(
|
|
|
biz_dt, name, _TARGET_STRATEGIES
|
|
biz_dt, name, _TARGET_STRATEGIES
|
|
|
)
|
|
)
|
|
|
- rows.append(_build_stats_row(category_id, name, biz_dt, matches))
|
|
|
|
|
|
|
+ real_matches = pool_repo.list_real_metrics_by_name_like(biz_dt, name)
|
|
|
|
|
+ rows.append(
|
|
|
|
|
+ _build_stats_row(category_id, name, biz_dt, matches, real_matches)
|
|
|
|
|
+ )
|
|
|
|
|
|
|
|
if len(rows) >= _STATS_UPSERT_BATCH:
|
|
if len(rows) >= _STATS_UPSERT_BATCH:
|
|
|
upserted += stats_repo.upsert_rows(rows)
|
|
upserted += stats_repo.upsert_rows(rows)
|
|
@@ -304,13 +444,17 @@ def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> d
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
classify_stats = _classify_words(partition_date)
|
|
classify_stats = _classify_words(partition_date)
|
|
|
|
|
+ real_metric_stats = enrich_real_rov_vov_7d(partition_date)
|
|
|
popularity_stats = compute_popularity_stats(partition_date)
|
|
popularity_stats = compute_popularity_stats(partition_date)
|
|
|
|
|
+ tree_weight_stats = compute_category_tree_weight(partition_date)
|
|
|
|
|
|
|
|
result = {
|
|
result = {
|
|
|
"partition_date": partition_date,
|
|
"partition_date": partition_date,
|
|
|
**sync_stats,
|
|
**sync_stats,
|
|
|
|
|
+ "real_metrics": real_metric_stats,
|
|
|
"classify": classify_stats,
|
|
"classify": classify_stats,
|
|
|
"popularity": popularity_stats,
|
|
"popularity": popularity_stats,
|
|
|
|
|
+ "tree_weight": tree_weight_stats,
|
|
|
"synced_at": datetime.now().isoformat(),
|
|
"synced_at": datetime.now().isoformat(),
|
|
|
}
|
|
}
|
|
|
logger.info("Multi demand pool sync completed: %s", result)
|
|
logger.info("Multi demand pool sync completed: %s", result)
|