mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
ThearchyHelios
330 lines
20 KiB
Python
330 lines
20 KiB
Python
"""波动基线、趋势（app/analysis/series.py，细案 01 §3、§4）：常规 / 偏离 / 明显偏离、σ̂ = 0、历史不够、本期不完整、缺期、日数据
|
||
lag 7、新高新低；上升 / 下降 / 无趋势、连续期数、缺期。每个用例都顺带核对复算和运行逐字段相同。"""
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
from datetime import date, timedelta
|
||
from fractions import Fraction
|
||
|
||
import pytest
|
||
|
||
from app.analysis.series import MAD_SCALE, run_length
|
||
from tests.analysis_samples import analyze, call_error, fixed_prober, snap
|
||
|
||
SQL = "SELECT DATE(order_date, 'weekday 0', '-6 days') AS week, COUNT(*) AS orders FROM orders GROUP BY 1"
|
||
#: 每周一的日期：最后一周是 2026-10-12（到 10-18，运行当天 10-21 之前已经过完）
|
||
LAST_MONDAY = date(2026, 10, 12)
|
||
|
||
|
||
def weeks(values, *, last=LAST_MONDAY):
|
||
start = last - timedelta(days=7 * (len(values) - 1))
|
||
return [[(start + timedelta(days=7 * i)).isoformat(), v] for i, v in enumerate(values)]
|
||
|
||
|
||
def series(rows, sql=SQL, cols=("week", "orders")):
|
||
return snap("Q1", sql, list(cols), rows, types={cols[0]: "text", cols[1]: "number"})
|
||
|
||
|
||
async def baseline(rows, **params):
|
||
params = {"query": "Q1", "period": "week", "measure": "orders", "label": "订单数", **params}
|
||
return await analyze("baseline", snaps=[series(rows)], **params)
|
||
|
||
|
||
async def trend(rows, **params):
|
||
params = {"query": "Q1", "period": "week", "measure": "orders", "label": "线上销售额", **params}
|
||
return await analyze("trend", snaps=[series(rows)], **params)
|
||
|
||
|
||
NOISE = [100, 102, 98, 101, 99, 103, 100, 97, 102, 101, 99, 100, 98, 103, 101, 100, 99, 102, 100]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 波动基线
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_level_basis_with_hand_computed_median_and_mad():
|
||
rows = [[i + 1, v] for i, v in enumerate([10, 12, 11, 13, 9, 10, 12, 11, 16])]
|
||
result = await baseline(rows, basis="level")
|
||
# 历史中位数 11，MAD 1，σ̂ = 1.4826，z = (16 − 11) / 1.4826
|
||
values = result["values"]
|
||
assert values["median"] == "11" and values["n_history"] == "8" and values["rank"] == "1"
|
||
assert Fraction(values["z"]) == pytest.approx(Fraction(5) / MAD_SCALE)
|
||
assert values["band_lo"] == "8.0348" and values["band_hi"] == "13.9652"
|
||
assert result["verdict"] == {"code": "far_outside_band", "subjects": [], "direction": "up"}
|
||
assert result["phrase"].startswith("订单数本期为 16，明显超出过去 8 期的常规波动范围（8.03 至 13.97）")
|
||
assert any(c["code"] == "record_high" for c in result["caveats"])
|
||
assert result["params"]["grain"] == "seq" and result["params"]["lag"] == 1
|
||
assert result["settings"] == {"k_warn": 2.0, "k_alert": 3.0, "min_history": 8, "max_history": 52}
|
||
assert result["table"]["rows"][-1][-1] == "本期"
|
||
assert result["table"]["column_types"] == {"week": "text", "value": "number", "change": "number", "role": "text"}
|
||
|
||
|
||
async def test_change_basis_within_outside_and_far_outside():
|
||
probe = fixed_prober({("orders", "order_date"): ("2025-01-01", "2026-10-20")})
|
||
within = await analyze("baseline", snaps=[series(weeks(NOISE + [101]))], probe=probe, query="Q1", period="week",
|
||
measure="orders", label="订单数")
|
||
assert within["verdict"]["code"] == "within_band"
|
||
assert within["phrase"].startswith("订单数本期变化 +1.0%，在过去 18 周的常规波动范围内（")
|
||
assert within["params"]["today"] == "2026-10-21" and within["params"]["grain"] == "week"
|
||
assert within["probes"][0]["table"] == "orders" and within["probes"][0]["column"] == "order_date"
|
||
assert not any(c["code"] == "comparability" for c in within["caveats"])
|
||
outside = await analyze("baseline", snaps=[series(weeks(NOISE + [107]))], probe=probe, query="Q1",
|
||
period="week", measure="orders")
|
||
assert outside["verdict"]["code"] == "outside_band" and "超出过去 18 周" in outside["phrase"]
|
||
far = await analyze("baseline", snaps=[series(weeks(NOISE + [70]))], probe=probe, query="Q1", period="week",
|
||
measure="orders", label="订单数")
|
||
assert far["verdict"] == {"code": "far_outside_band", "subjects": [], "direction": "down"}
|
||
assert far["phrase"].startswith("订单数本期变化 -30.0%，明显超出过去 18 周的常规波动范围")
|
||
assert any(c["code"] == "record_low" and c["text"] == "本期的订单数是过去 18 周里最低的" for c in far["caveats"])
|
||
|
||
|
||
async def test_without_a_table_range_completeness_is_said_to_be_unverified_but_not_in_the_phrase():
|
||
result = await baseline(weeks(NOISE + [101]))
|
||
note = next(c for c in result["caveats"] if c["code"] == "comparability")
|
||
assert note["text"] == "没能核对本期的数据是否到齐（查不到表里这一列的日期范围）"
|
||
assert note["text"] not in result["phrase"]
|
||
|
||
|
||
async def test_flat_history_has_no_spread():
|
||
rows = [[i + 1, 50] for i in range(10)]
|
||
same = await baseline(rows, basis="level")
|
||
assert same["verdict"]["code"] == "within_band" and same["values"]["z"] is None
|
||
assert same["phrase"] == "订单数本期为 50，和过去 9 期一样（每期都是 50）"
|
||
jump = await baseline(rows[:-1] + [[10, 51]], basis="level")
|
||
assert jump["verdict"] == {"code": "far_outside_band", "subjects": [], "direction": "up"}
|
||
assert jump["phrase"].startswith("订单数本期为 51，明显超出过去 9 期的常规波动范围（每期都是 50）")
|
||
|
||
|
||
async def test_short_history_partial_target_gaps_and_zero_bases_are_refused():
|
||
short = await baseline(weeks([100, 101, 99, 102, 98, 100]))
|
||
# 6 周：第一周没有上一周可比，本期不算历史，历史的变化只有 4 个
|
||
assert short["reason_code"] == "history_short" and short["reason"] == "只有 4 期历史，至少要 8 期"
|
||
assert short["phrase"] == "波动基线没有做成：只有 4 期历史，至少要 8 期"
|
||
# 最后一周是 10-19 到 10-25：运行当天 10-21 还没过完
|
||
partial = await baseline(weeks(NOISE + [60], last=date(2026, 10, 19)))
|
||
assert partial["reason_code"] == "partial_target"
|
||
assert partial["reason"] == "本期数据只到 2026-10-20，还差 5 天，和完整的周比没有意义"
|
||
# 周过完了，但表里的数据只到 10-17
|
||
probe = fixed_prober({("orders", "order_date"): ("2025-01-01", "2026-10-17")})
|
||
late = await analyze("baseline", snaps=[series(weeks(NOISE + [60]))], probe=probe, query="Q1", period="week",
|
||
measure="orders")
|
||
assert late["reason"] == "本期数据只到 2026-10-17，还差 1 天，和完整的周比没有意义"
|
||
gap_rows = weeks(NOISE + [100, 100])
|
||
del gap_rows[-2] # 本期的上一周没有
|
||
gap = await baseline(gap_rows)
|
||
assert gap["reason_code"] == "target_gap"
|
||
zero = await baseline(weeks(NOISE + [0, 10]))
|
||
assert zero["reason_code"] == "zero_base"
|
||
unreadable = await baseline([["上周", 1], ["本周", 2]])
|
||
assert unreadable["reason_code"] == "period_unreadable" and "上周" in unreadable["reason"]
|
||
|
||
|
||
async def test_daily_series_compare_with_the_same_weekday_last_week():
|
||
start = date(2026, 7, 1)
|
||
rows = [[(start + timedelta(days=i)).isoformat(), 100 + (i % 7) * 10] for i in range(100)]
|
||
result = await analyze("baseline", snaps=[series(rows, cols=("day", "orders"))], query="Q1", period="day",
|
||
measure="orders", label="订单数", check=True)
|
||
assert result["params"]["grain"] == "day" and result["params"]["lag"] == 7
|
||
# 星期效应：和上周同一天比，每天的变化都是 0
|
||
assert result["values"]["median"] == "0" and result["verdict"]["code"] == "within_band"
|
||
assert result["phrase"].startswith("订单数比上周同一天变化 0.0%，和过去 52 天一样")
|
||
yoy = await baseline(weeks(NOISE * 3), lag=52) # 57 周：能和 52 周前比的历史只有 4 期
|
||
assert yoy["reason_code"] == "history_short" and yoy["reason"] == "只有 4 期历史，至少要 8 期"
|
||
assert (await baseline(weeks(NOISE * 4), lag=52))["status"] == "ok"
|
||
assert (await call_error("baseline", snaps=[series(weeks(NOISE))], query="Q1", period="week", measure="orders",
|
||
lag=0)).code == "bad_lag"
|
||
|
||
|
||
async def test_duplicate_periods_and_given_targets():
|
||
rows = weeks(NOISE + [101])
|
||
error = await call_error("baseline", snaps=[series(rows + [rows[-1]])], query="Q1", period="week",
|
||
measure="orders")
|
||
assert error.code == "duplicate_rows" and "在 SQL 里按 week 汇总" in error.message
|
||
earlier = await baseline(rows, target=rows[-3][0])
|
||
assert earlier["params"]["target"] == rows[-3][0] and earlier["values"]["n_history"] == "16"
|
||
missing = await call_error("baseline", snaps=[series(rows)], query="Q1", period="week", measure="orders",
|
||
target="2020-01-06")
|
||
assert missing.code == "period_not_found"
|
||
|
||
|
||
async def test_iso_week_labels_and_months():
|
||
labels = [[f"2026-W{w:02d}", v] for w, v in zip(range(22, 42), NOISE + [101])]
|
||
result = await baseline(labels)
|
||
assert result["params"]["grain"] == "week" and result["verdict"]["code"] == "within_band"
|
||
months = [[f"2025-{m:02d}", 100 + m] for m in range(1, 13)] + [[f"2026-{m:02d}", 112 + m] for m in range(1, 10)]
|
||
monthly = await baseline(months)
|
||
assert monthly["params"]["grain"] == "month" and "过去 19 个月" in monthly["phrase"]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 趋势
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_rising_and_falling_trends_with_run_length():
|
||
rising = await trend(weeks([100 + 3 * i + (i % 3) for i in range(12)]))
|
||
assert rising["verdict"] == {"code": "rising", "subjects": [], "direction": "up"}
|
||
assert rising["values"]["n"] == "12" and rising["values"]["run_length"] == "11"
|
||
assert rising["phrase"].startswith("过去 12 周线上销售额呈上升趋势（每周约 +")
|
||
assert "，已连续 11 周上升" in rising["phrase"]
|
||
assert rising["settings"] == {"trend_min_points": 6, "trend_alpha": 0.05, "max_history": 52}
|
||
assert rising["table"]["column_types"] == {"week": "text", "value": "number", "change": "number"}
|
||
falling = await trend(weeks([200, 196, 197, 190, 185, 186, 180, 181, 175, 170]))
|
||
assert falling["verdict"]["code"] == "falling"
|
||
assert falling["values"]["run_length"] == "2" and "已连续" not in falling["phrase"]
|
||
|
||
|
||
async def test_no_trend_and_exact_small_sample_p():
|
||
flat = await trend(weeks([100, 104, 98, 103, 97, 101, 99, 102]))
|
||
assert flat["verdict"] == {"code": "no_trend", "subjects": [], "direction": None}
|
||
assert flat["phrase"] == "过去 8 周线上销售额没有明显的上升或下降趋势"
|
||
six = await trend(weeks([1, 2, 3, 4, 5, 6]))
|
||
# 没有并列、n ≤ 10：精确分布，单调的 6 期双侧 p = 2 / 6!
|
||
assert Fraction(six["values"]["p"]) == pytest.approx(Fraction(2, 720))
|
||
assert six["verdict"]["code"] == "rising" and six["values"]["slope"] == "1"
|
||
|
||
|
||
def test_run_length_counts_consecutive_changes_from_the_end():
|
||
assert run_length([Fraction(v) for v in (1, 2, 3, 2, 3, 4, 5)]) == (3, "up")
|
||
assert run_length([Fraction(v) for v in (5, 4, 3)]) == (2, "down")
|
||
assert run_length([Fraction(v) for v in (1, 2, 2)]) == (0, None)
|
||
|
||
|
||
async def test_trend_needs_enough_consecutive_points():
|
||
rows = weeks([100 + i for i in range(10)])
|
||
del rows[-4] # 最后连着的只有 3 期
|
||
gap = await trend(rows)
|
||
assert gap["reason_code"] == "series_gap" and gap["reason"] == "最后连着的只有 3 期，再往前缺期（至少要连着的 6 期）"
|
||
short = await trend(weeks([1, 2, 3, 4]))
|
||
assert short["reason_code"] == "history_short" and short["reason"] == "只有 4 期，至少要 6 期"
|
||
partial = await trend(weeks([100 + i for i in range(10)], last=date(2026, 10, 19)))
|
||
assert partial["reason_code"] == "partial_target"
|
||
|
||
|
||
async def test_daily_trends_warn_about_weekday_effects():
|
||
start = date(2026, 8, 1)
|
||
rows = [[(start + timedelta(days=i)).isoformat(), 100 + i] for i in range(30)]
|
||
result = await analyze("trend", snaps=[series(rows, cols=("day", "orders"))], query="Q1", period="day",
|
||
measure="orders")
|
||
assert result["verdict"]["code"] == "rising"
|
||
assert any(c["code"] == "weekday_effect" and c["text"] == "日数据有星期效应，看趋势建议按周汇总"
|
||
for c in result["caveats"])
|
||
assert "日数据有星期效应" in result["phrase"]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 周号按 SQL 的算法算起止（SQLite 的 strftime('%W') 不是 ISO 周）
|
||
# --------------------------------------------------------------------------
|
||
|
||
#: SQLite 按 %W 写的周：第 1 周从这一年第一个周一起，之前的几天是第 0 周，跨年那一周被切成两期
|
||
WEEKNO = "SELECT strftime('%Y-W%W', order_date) AS week, COUNT(*) AS orders FROM orders GROUP BY week ORDER BY week"
|
||
|
||
|
||
def sqlite_weeks(first, last, sql=WEEKNO):
|
||
"""真用 SQLite 按 sql 汇总：每天 100 笔订单，从 first 到 last（含）。"""
|
||
db = sqlite3.connect(":memory:")
|
||
db.execute("CREATE TABLE orders (order_date TEXT)")
|
||
day = first
|
||
while day <= last:
|
||
db.executemany("INSERT INTO orders VALUES (?)", [(day.isoformat(),)] * 100)
|
||
day += timedelta(days=1)
|
||
rows = [list(r) for r in db.execute(sql)]
|
||
db.close()
|
||
return rows
|
||
|
||
|
||
def weekly(rows, sql=WEEKNO, **kw):
|
||
return snap("Q1", sql, ["week", "orders"], rows, types={"week": "text", "orders": "number"}, **kw)
|
||
|
||
|
||
async def test_a_sqlite_week_that_is_not_over_is_not_a_whole_week():
|
||
# 运行当天 10-21、数据到 10-20：SQLite 的 2026-W42 是 10-19 到 10-25，只有两天的数据（ISO 的 W42 是 10-12 到 10-18）
|
||
rows = sqlite_weeks(date(2026, 6, 1), date(2026, 10, 20))
|
||
assert rows[-1] == ["2026-W42", 200]
|
||
probe = fixed_prober({("orders", "order_date"): ("2026-06-01", "2026-10-20")})
|
||
result = await analyze("baseline", snaps=[weekly(rows)], probe=probe, query="Q1", period="week", measure="orders",
|
||
label="订单数")
|
||
assert result["reason_code"] == "partial_target"
|
||
assert result["reason"] == "本期数据只到 2026-10-20，还差 5 天，和完整的周比没有意义"
|
||
assert result["params"]["target"] == "2026-W42"
|
||
whole = await analyze("baseline", snaps=[weekly(rows)], probe=probe, query="Q1", period="week", measure="orders",
|
||
label="订单数", target="2026-W41")
|
||
assert whole["status"] == "ok" and whole["verdict"]["code"] == "within_band"
|
||
# 不知道方言（认不出算法）：起止为空，本期齐不齐核对不了，绝不判齐
|
||
unknown = await analyze("baseline", snaps=[weekly(rows)], query="Q1", period="week", measure="orders",
|
||
label="订单数", target="2026-W41")
|
||
note = next(c["text"] for c in unknown["caveats"] if c["code"] == "comparability")
|
||
assert note == "没能核对本期的数据是否到齐（期次写成「年-W周号」，认不出 SQL 是按哪种周号算的，不知道是哪几天）"
|
||
|
||
|
||
async def test_sqlite_weeks_across_new_year_are_readable_and_the_cut_weeks_are_left_out():
|
||
rows = sqlite_weeks(date(2025, 9, 29), date(2026, 3, 1))
|
||
assert ["2025-W52", 300] in rows and ["2026-W00", 400] in rows
|
||
probe = fixed_prober({("orders", "order_date"): ("2025-09-29", "2026-03-01")})
|
||
ask = {"snaps": [weekly(rows)], "probe": probe, "query": "Q1", "period": "week", "measure": "orders",
|
||
"label": "订单数"}
|
||
trend = await analyze("trend", **ask)
|
||
# 最后连着的整周是 2026-W01 到 W08：再往前的 2026-W00 只有 4 天
|
||
assert trend["status"] == "ok" and trend["values"]["n"] == "8" and trend["verdict"]["code"] == "no_trend"
|
||
assert [r[0] for r in trend["table"]["rows"]][0] == "2026-W01"
|
||
year_end = await analyze("baseline", **ask, target="2026-W00")
|
||
assert year_end["reason_code"] == "partial_target"
|
||
assert year_end["reason"] == "本期只有 2026-01-01 到 2026-01-04 这 4 天，不是整周，和完整的周比没有意义"
|
||
after = await analyze("baseline", **ask, target="2026-W01")
|
||
assert after["reason_code"] == "target_gap"
|
||
assert after["reason"] == "比较期（2026-W00）只有 4 天，不是整周，和它比没有意义"
|
||
later = await analyze("baseline", **ask, target="2026-W03")
|
||
# 历史：2025-W40 到 W51 各和上一周比（12 个），2026-W02 和 W01 比（1 个）；W01 和 W00、W00 和 2025-W52 都不算
|
||
assert later["status"] == "ok" and later["values"]["n_history"] == "13"
|
||
assert {"code": "partial_periods", "text": "有 2 期不到整周（跨年被切开），没有算进历史"} in later["caveats"]
|
||
assert "跨年" not in later["phrase"]
|
||
short = await analyze("trend", **{**ask, "snaps": [weekly([r for r in rows if r[0] <= "2026-W03"])]})
|
||
assert short["reason_code"] == "series_gap"
|
||
assert short["reason"] == "最后连着的只有 3 期，再往前的 2026-W00 只有 4 天，不是整周（至少要连着的 6 期）"
|
||
|
||
|
||
async def test_unreadable_periods_name_the_values_that_cannot_be_read():
|
||
rows = [[f"2026-W{w:02d}", 100] for w in range(30, 42)] + [["上周", 100]]
|
||
result = await baseline(rows)
|
||
assert result["reason_code"] == "period_unreadable"
|
||
assert result["reason"] == "期次列「week」的值认不出是哪一期（上周）：要日期、YYYY-MM、YYYY-Www、YYYY-Qn 或整数序号"
|
||
mixed = await baseline([[f"2026-W{w:02d}", 100] for w in range(30, 42)] + [["2026-10-12", 100]])
|
||
assert mixed["reason"] == "期次列「week」的写法不一：2026-10-12 和其余的不是同一种写法"
|
||
missing = await analyze("baseline", snaps=[weekly([["2024-W00", 1], ["2024-W01", 2]])], probe=fixed_prober(),
|
||
query="Q1", period="week", measure="orders")
|
||
assert missing["reason"] == "期次列「week」里的 2024-W00 按 SQL 里的周号算法没有这一周"
|
||
|
||
|
||
async def test_an_empty_comparison_period_is_said_as_empty_not_as_a_gap():
|
||
rows = weeks(NOISE + [100, 101])
|
||
rows[-2][1] = None # 本期的上一周有这一行，值为空
|
||
result = await baseline(rows)
|
||
assert result["reason_code"] == "target_gap"
|
||
assert result["reason"] == f"比较期（{rows[-2][0]}）的订单数为空，没有比较期的值"
|
||
tail = weeks([100 + i for i in range(10)])
|
||
tail[-4][1] = None
|
||
broken = await trend(tail)
|
||
assert broken["reason"] == f"最后连着的只有 3 期，再往前的 {tail[-4][0]} 那一期的线上销售额为空（至少要连着的 6 期）"
|
||
|
||
|
||
async def test_sequence_periods_say_completeness_cannot_be_checked():
|
||
rows = [[i + 1, v] for i, v in enumerate([10, 12, 11, 13, 9, 10, 12, 11, 16])]
|
||
result = await baseline(rows, basis="level")
|
||
assert {"code": "comparability", "text": "没能核对本期的数据是否到齐（期次不是日期，不知道是哪几天）"} \
|
||
in result["caveats"]
|
||
|
||
|
||
async def test_a_falling_trend_whose_last_weeks_rise_says_but():
|
||
falling = await trend(weeks([200, 196, 190, 188, 181, 176, 170, 165, 160, 162, 165, 168]))
|
||
assert falling["verdict"]["code"] == "falling" and falling["values"]["run_direction"] == "up"
|
||
assert falling["phrase"].endswith("，但最近已连续 3 周上升")
|
||
|
||
|
||
async def test_a_period_column_named_like_a_result_column_is_a_call_error():
|
||
rows = [[r[0], r[1]] for r in weeks(NOISE + [101])]
|
||
named = snap("Q1", "SELECT DATE(order_date, 'weekday 0', '-6 days') AS value, COUNT(*) AS orders FROM orders "
|
||
"GROUP BY 1", ["value", "orders"], rows)
|
||
error = await call_error("baseline", snaps=[named], query="Q1", period="value", measure="orders")
|
||
assert error.code == "reserved_column" and "期次列「value」和结果表的列重名" in error.message
|