mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 01:11:59 +00:00
周号按 SQL 里的格式串算真实起止（复核 R3）：「YYYY-Www」不再一律当 ISO 周。periods.week_scheme 看期次列在最外层 SELECT 里的写法（从 CTE、子查询里原样取出的往里找），认格式化函数里配套的年份和周号——SQLite strftime 的 %Y-%W / %Y-%U / %G-%V、MySQL DATE_FORMAT 的 %Y-%U / %Y-%u / %X-%V / %x-%v、PostgreSQL / Oracle to_char 的 YYYY-WW / IYYY-IW；week_span 按算法算一周的起止（W00、W53 按各自的算法，不走 fromisocalendar），先后按算法排（跨年切开的两期也连着）。认不出算法（方言不知道、只取了现成的字符串、年份和周号不配套）的起止置空，本期齐不齐一律核对不了、附说明，绝不判齐，先后按周号排。按年截断的算法里跨年切开的那两期不到整周：当本期判 partial_target，当比较期判 target_gap 并说清只有几天，不算进波动基线的历史（附 partial_periods 说明，只在证据面板），趋势取最后连着的整周。可比性检查按期次分组时也走这套：有一期不到整周的按实际天数比窗口、照算星期构成。period_unreadable 的原因里举认不出的那几个值（写法认不出、写法混着的少数、算法里没有的那一周、对不上给的粒度分开说）；比较期那一格为空时照实说为空，不再说成缺期（趋势往前断开的原因同样分开说）。期次是序号、周号认不出算法时也附一句核对不了。契约 CAVEATS 加 partial_periods
This commit is contained in:
ThearchyHelios
@@ -70,7 +70,8 @@ class Span:
|
||||
table: str | None = None
|
||||
column: str | None = None
|
||||
sql_window: Any = None
|
||||
unit: str | None = None # 这一段是不是整月、整季、整年（calendar_unit），按期次时是粒度
|
||||
unit: str | None = None # 这一段是不是整月、整季、整年（calendar_unit），按期次时是粒度（不是整期的为 None）
|
||||
short: int | None = None # 按期次的这一期不到整期时有几天（periods.short_days：跨年被切开的周）
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -238,7 +239,9 @@ def period_span(label: str, snap: Snap, period: Any, item: pd.Period, grain: str
|
||||
"""按期次分组的一份快照里的一期：这一期覆盖的日子，加上 SQL 在期次那一列上的时间条件。"""
|
||||
from app.data import coverage
|
||||
|
||||
span = Span(label, snap, item.start, item.end, unit=grain)
|
||||
span = Span(label, snap, item.start, item.end, unit=grain, short=pd.short_days(item, grain))
|
||||
if span.short:
|
||||
span.unit = None
|
||||
found = date_source(snap.sql, period)
|
||||
if found is None:
|
||||
return span
|
||||
@@ -268,10 +271,13 @@ def _range(span: Span, book: Book) -> tuple[bool, date | None, date | None, bool
|
||||
|
||||
|
||||
def span_coverage(span: Span, today: date, book: Book) -> tuple[str, str | None]:
|
||||
"""这一段（窗口或一期）的数据齐不齐：(pass / short / unknown, 不齐的那一句)。没过完（终点不早于运行当天：当天的数据还没
|
||||
到齐）、SQL 的时间条件截掉了一部分、表里的数据没到（最晚早于终点）或来得晚（最早晚于起点），都算不齐。"""
|
||||
"""这一段（窗口或一期）的数据齐不齐：(pass / short / unknown, 不齐的那一句)。这一期本身不到整期（跨年被切开的周）、没过完
|
||||
（终点不早于运行当天：当天的数据还没到齐）、SQL 的时间条件截掉了一部分、表里的数据没到（最晚早于终点）或来得晚（最早晚于
|
||||
起点），都算不齐。"""
|
||||
if span.start is None or span.end is None:
|
||||
return "unknown", None
|
||||
if span.short:
|
||||
return "short", f"{span.label}只有 {span.start.isoformat()} 到 {span.end.isoformat()} 这 {span.short} 天，不是整周"
|
||||
lo, hi = span.start, span.end
|
||||
window = span.sql_window
|
||||
if window is not None:
|
||||
@@ -417,20 +423,32 @@ def compare_queries(base: Snap, target: Snap, measure: str | None, facts: Facts,
|
||||
|
||||
def compare_periods(snap: Snap, period: str, base: pd.Period, target: pd.Period, grain: str, measure: str | None,
|
||||
facts: Facts, book: Book, today: date) -> Outcome:
|
||||
"""同一份快照里按期次分组的两期：天然同口径；窗口按期次的粒度；整期看这两期各自齐不齐。"""
|
||||
"""同一份快照里按期次分组的两期：天然同口径；窗口按期次的粒度（有一期不到整期的按实际天数比，星期构成也照算）；整期看
|
||||
这两期各自齐不齐。期次不知道是哪几天的（序号、认不出算法的周号）窗口、整期、星期构成都核对不了。"""
|
||||
same = Check("same_query", status="pass")
|
||||
b_span = period_span("基期", snap, period, base, grain, measure, facts, book)
|
||||
t_span = period_span("本期", snap, period, target, grain, measure, facts, book)
|
||||
flow = bool(measure) and facts.measure(snap, measure)[0] == "flow" # type: ignore[arg-type]
|
||||
if grain == "seq":
|
||||
whole = not (b_span.short or t_span.short)
|
||||
if grain == "seq" or base.start is None or target.start is None:
|
||||
checks = [same, Check("window"), Check("complete"), Check("weekdays"), _data_check([("", snap)], measure)]
|
||||
else:
|
||||
checks = [same, _window_check(b_span, t_span, flow, grouped=True),
|
||||
checks = [same, _window_check(b_span, t_span, flow, grouped=whole),
|
||||
_complete_check([b_span, t_span], today, book, flow),
|
||||
_weekday_check(b_span, t_span, grain=grain), _data_check([("", snap)], measure)]
|
||||
_weekday_check(b_span, t_span, grain=grain if whole else None), _data_check([("", snap)], measure)]
|
||||
return _decide(checks, b_span, t_span)
|
||||
|
||||
|
||||
def period_complete(snap: Snap, period: str, item: pd.Period, grain: str, measure: str | None, facts: Facts,
|
||||
book: Book, today: date) -> tuple[str, str | None, str | None]:
|
||||
"""按期次分组的一份快照里本期齐不齐（波动基线、趋势、单份快照的变化拆解共用）：(pass / short / unknown, 不齐的那一句,
|
||||
核对不了时附带说明的那一句)。期次没有日期（序号）、周号认不出算法的，起止不知道，一律核对不了，不判齐。"""
|
||||
if item.start is None:
|
||||
return "unknown", None, ph.UNKNOWN_WEEKS if grain == "week" else ph.NO_DATES
|
||||
status, why = span_coverage(period_span("本期", snap, period, item, grain, measure, facts, book), today, book)
|
||||
return status, why, ph.UNVERIFIED_COMPLETE if status == "unknown" else None
|
||||
|
||||
|
||||
def table_of(outcome: Outcome) -> dict[str, Any]:
|
||||
"""检查表（结论的 table）：检查项、结果、基期、本期、说明。"""
|
||||
rows = [[CHECK_NAMES[c.id], STATUS_NAMES[c.status], c.base, c.target, "；".join(c.reasons) or None]
|
||||
@@ -513,7 +531,7 @@ def comparable(raw: Mapping[str, Any], env: Any) -> Any:
|
||||
raise CallError("period_not_found", f"期次列「{period}」里没有「{params[side]}」：{what}要写 {target.ref} "
|
||||
"里有的期次值")
|
||||
env.inputs.use(target, "data", [period, *([measure] if measure else [])], rows["base"] + rows["target"])
|
||||
parsed = pd.parse(values, grain)
|
||||
parsed = pd.parse(values, grain, weeks=pd.week_scheme(target.sql, period, target.dialect))
|
||||
if parsed is None:
|
||||
item_b = item_t = pd.Period(None, 0, None, None)
|
||||
grain = "seq"
|
||||
@@ -538,5 +556,5 @@ def _only_number(snap: Snap) -> str | None:
|
||||
|
||||
|
||||
__all__ = ["CHECKS", "CHECK_NAMES", "Check", "Outcome", "STATUS_NAMES", "Span", "comparable", "compare_periods",
|
||||
"compare_queries", "period_span", "reasons", "same_query", "span_coverage", "sql_span", "table_of",
|
||||
"values_of"]
|
||||
"compare_queries", "period_complete", "period_span", "reasons", "same_query", "span_coverage", "sql_span",
|
||||
"table_of", "values_of"]
|
||||
|
||||
@@ -92,7 +92,7 @@ REASONS: tuple[str, ...] = (
|
||||
CAVEATS: tuple[str, ...] = (
|
||||
"offset", "simpson", "small_denominator", "record_high", "record_low", "weekday_effect",
|
||||
"multiple_comparisons", "outlier_sensitive", "small_sample", "measure_unlabeled", "null_values",
|
||||
"headline_basis", "comparability",
|
||||
"headline_basis", "comparability", "partial_periods",
|
||||
)
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
+293
-18
@@ -6,26 +6,32 @@
|
||||
且相邻两期隔整月 → month（月份都在 1、4、7、10 月且隔整季 → quarter）；相邻两期都隔 7 天的整数倍 → week（这个日期当
|
||||
一周的第一天）；否则 day。
|
||||
- 月份：'YYYY-MM'、'YYYYMM'（6 位，月份 01–12）→ month。
|
||||
- ISO 周：'YYYY-Www'（2026-W42 是 2026-10-12 到 10-18）→ week。SQLite 的 strftime('%W') 不是 ISO 周号（年初那几天
|
||||
算第 0 周），按它写的期次覆盖的日子会差一周：方法里要求期次写成每周第一天的日期。
|
||||
- 周号：'YYYY-Www' → week。同样写成 2026-W42，不同的数据库、不同的格式串算出来的是不同的日子（ISO 周是 10-12 到 10-18，
|
||||
SQLite 的 strftime('%Y-W%W') 是 10-19 到 10-25），所以起止按 SQL 里写这一列的格式串算（week_scheme 认出周号算法，见
|
||||
WEEK_SCHEMES）；认不出算法的（方言不知道、只取了一列现成的字符串、年份和周号不配套）起止为空，本期齐不齐核对不了，先后按
|
||||
周号排（同一年里周号连着的算连着，上一年的第 52、53 周和下一年的第 0、1 周算连着）。按年截断的周号算法里，跨年那一周
|
||||
被切成两期，年末、年初那两期不到 7 天（short_days）：不是整周，不能当本期、比较期，也不算进历史。
|
||||
- 季度：'YYYY-Qn'、'YYYYQn' → quarter。
|
||||
- 整数序号（不是上面几种的整数）→ seq：能排序、能判连续，不知道覆盖哪几天（本期完不完整核对不了）。
|
||||
|
||||
带时刻（非零点）的、写法混着的、认不出的都算认不出（period_unreadable）。
|
||||
|
||||
每一期记 ordinal：同一粒度里相邻两期的 ordinal 差 1（日：天数，周：从第一期起第几个 7 天，月：年 × 12 + 月，季：年 × 4 +
|
||||
季，序号：本身），据此排序、判缺期、按 lag 找比较期。start / end 是这一期覆盖的第一天和最后一天（含），seq 没有。
|
||||
每一期记 ordinal：同一粒度里相邻两期的 ordinal 差 1（日：天数，周：日期写的是从第一期起第几个 7 天、周号按它的算法排，
|
||||
月：年 × 12 + 月，季：年 × 4 + 季，序号：本身），据此排序、判缺期、按 lag 找比较期。start / end 是这一期覆盖的第一天和最后
|
||||
一天（含），seq 和认不出算法的周号没有。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import calendar
|
||||
import re
|
||||
from collections.abc import Sequence
|
||||
from collections.abc import Iterable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta
|
||||
from functools import lru_cache
|
||||
from typing import Any
|
||||
|
||||
from app.data import coverage
|
||||
from app.data.names import name_key
|
||||
|
||||
GRAINS = ("day", "week", "month", "quarter")
|
||||
#: 粒度的叫法：数量词（过去 12 周、12 个月）、每…（每周、每月）
|
||||
@@ -38,6 +44,32 @@ _MONTH = re.compile(r"^(\d{4})-?(\d{2})$")
|
||||
_WEEK = re.compile(r"^(\d{4})-?W(\d{1,2})$", re.I)
|
||||
_QUARTER = re.compile(r"^(\d{4})-?Q([1-4])$", re.I)
|
||||
|
||||
#: 「YYYY-Www」的周号算法（week_scheme 按 SQL 认，week_span 按它算一周的起止）：
|
||||
#: - iso：ISO 8601。周一开始，第 1 周是有 4 天以上落在这一年的那周，年份是 ISO 年（年初、年末的几天算到相邻的年）
|
||||
#: - monday：周一开始，第 1 周从这一年第一个周一起，之前的几天是第 0 周，按年截断（SQLite、C 的 %W）
|
||||
#: - sunday：周日开始，其余同 monday（%U；MySQL WEEK 模式 0）
|
||||
#: - monday4：周一开始，第 1 周是有 4 天以上落在这一年的那周，之前的几天是第 0 周，按年截断（MySQL 的 %u，WEEK 模式 1）
|
||||
#: - sunday_year：周日开始，第 1 周从这一年第一个周日起，年初之前的几天算上一年的最后一周（MySQL 的 %X 配 %V，WEEK 模式 2）
|
||||
#: - jan1：从 1 月 1 日起每 7 天一周，最后一周只有一两天（PostgreSQL、Oracle to_char 的 WW）
|
||||
WEEK_SCHEMES = ("iso", "monday", "sunday", "monday4", "sunday_year", "jan1")
|
||||
#: 各方言格式串里配套的（年份, 周号）写法 → 周号算法。不配套的（%Y 配 ISO 周号、IYYY 配 WW……）同一个标签会落在两段
|
||||
#: 日子上，认不出
|
||||
_WEEK_FORMATS: dict[str, dict[tuple[str, str], str]] = {
|
||||
"sqlite": {("%Y", "%W"): "monday", ("%Y", "%U"): "sunday", ("%G", "%V"): "iso"},
|
||||
"mysql": {("%Y", "%U"): "sunday", ("%Y", "%u"): "monday4", ("%X", "%V"): "sunday_year", ("%x", "%v"): "iso"},
|
||||
"postgres": {("YYYY", "WW"): "jan1", ("IYYY", "IW"): "iso"},
|
||||
"oracle": {("YYYY", "WW"): "jan1", ("IYYY", "IW"): "iso"},
|
||||
}
|
||||
#: 各方言格式串里年份、周号的写法（数格式串里有几个年份、几个周号用；方言不同，同一个 %u 意思不同）
|
||||
_YEAR_MARKS = {"sqlite": ("%Y", "%G"), "mysql": ("%Y", "%X", "%x"), "postgres": ("YYYY", "IYYY"),
|
||||
"oracle": ("YYYY", "IYYY")}
|
||||
_WEEK_MARKS = {"sqlite": ("%W", "%U", "%V"), "mysql": ("%U", "%u", "%V", "%v"), "postgres": ("WW", "IW"),
|
||||
"oracle": ("WW", "IW")}
|
||||
#: to_char 的格式串里先认长的：IYYY 里有 YYYY，IW、WW 都有 W（一个月里的第几周）
|
||||
_CHAR_MARKS = ("IYYY", "YYYY", "IW", "WW")
|
||||
#: 期次列从 CTE、子查询里原样取出时，最多往里找几层
|
||||
_NEST_MAX = 4
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Period:
|
||||
@@ -111,6 +143,219 @@ def _month_of(value: Any) -> date:
|
||||
return date(int(m.group(1)), int(m.group(2)), 1)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 周号：按算法算起止、排先后
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _week_label(value: Any) -> tuple[int, int]:
|
||||
m = _WEEK.match(str(value).strip())
|
||||
assert m is not None
|
||||
return int(m.group(1)), int(m.group(2))
|
||||
|
||||
|
||||
def _first_weekday(start: date, weekday: int) -> date:
|
||||
"""start 当天或之后的第一个星期几（周一 0 … 周日 6）。"""
|
||||
return start + timedelta(days=(weekday - start.weekday()) % 7)
|
||||
|
||||
|
||||
def week_span(scheme: str, year: int, week: int) -> tuple[date, date] | None:
|
||||
"""按周号算法，year 年第 week 周覆盖的第一天和最后一天（含）。这种算法里没有这一周（ISO 只有 52 周的年份写了第 53 周、
|
||||
1 月 1 日就是周一的年份写了第 0 周……）为 None。"""
|
||||
if scheme == "iso":
|
||||
try:
|
||||
start = date.fromisocalendar(year, week, 1)
|
||||
except ValueError:
|
||||
return None
|
||||
return start, start + timedelta(days=6)
|
||||
try:
|
||||
jan1, dec31 = date(year, 1, 1), date(year, 12, 31)
|
||||
except ValueError:
|
||||
return None
|
||||
if scheme == "jan1":
|
||||
if not 1 <= week <= 53:
|
||||
return None
|
||||
start = jan1 + timedelta(days=7 * (week - 1))
|
||||
return start, min(start + timedelta(days=6), dec31)
|
||||
if scheme == "sunday_year":
|
||||
start = _first_weekday(jan1, 6) + timedelta(days=7 * (week - 1))
|
||||
return (start, start + timedelta(days=6)) if week >= 1 and start.year == year else None
|
||||
if scheme == "monday":
|
||||
week1 = _first_weekday(jan1, 0)
|
||||
elif scheme == "sunday":
|
||||
week1 = _first_weekday(jan1, 6)
|
||||
elif scheme == "monday4":
|
||||
week1 = date.fromisocalendar(year, 1, 1) # ISO 第 1 周的周一，可能在上一年的 12 月
|
||||
else:
|
||||
return None
|
||||
if week == 0: # 第 1 周之前的那几天；第 1 周从 1 月 1 日（或更早）起就没有
|
||||
return (jan1, week1 - timedelta(days=1)) if week1 > jan1 else None
|
||||
start = week1 + timedelta(days=7 * (week - 1))
|
||||
if week < 0 or start > dec31:
|
||||
return None
|
||||
return max(start, jan1), min(start + timedelta(days=6), dec31)
|
||||
|
||||
|
||||
def _week_start(day: date, weekday: int) -> date:
|
||||
"""day 所在那一周（从星期几开始）的第一天。"""
|
||||
return day - timedelta(days=(day.weekday() - weekday) % 7)
|
||||
|
||||
|
||||
def _week_ordinal(scheme: str, span: tuple[date, date], year: int, base: int) -> int:
|
||||
"""一周在这种算法的先后里排第几（相邻两期差 1）。整 7 天一周的按那一周第一天的序数除以 7（同一个星期几开头的日子序数
|
||||
模 7 同余）；按年截断的，跨年那一周切成了两期，每过一个不是从这周第一天开始的元旦多一期（base 是这一列里最早的年份）。"""
|
||||
if scheme == "jan1":
|
||||
return year * 53 + (span[0] - date(year, 1, 1)).days // 7
|
||||
if scheme in ("iso", "sunday_year"):
|
||||
return span[0].toordinal() // 7
|
||||
weekday = 6 if scheme == "sunday" else 0
|
||||
splits = sum(1 for y in range(base + 1, year + 1) if date(y, 1, 1).weekday() != weekday)
|
||||
return _week_start(span[0], weekday).toordinal() // 7 + splits
|
||||
|
||||
|
||||
def _loose_ordinals(labels: Iterable[tuple[int, int]]) -> dict[tuple[int, int], int]:
|
||||
"""认不出周号算法时的先后：按（年，周号）排，同一年里周号连着的算连着；上一年的第 52、53 周和下一年的第 0、1 周算连着
|
||||
（核对不了跨年那里是不是缺了一周）；别的跨年当缺期（隔几期只是估计，至少隔一期）。"""
|
||||
out: dict[tuple[int, int], int] = {}
|
||||
prev: tuple[int, int] | None = None
|
||||
n = 0
|
||||
for year, week in sorted(set(labels)):
|
||||
if prev is not None:
|
||||
if year == prev[0]:
|
||||
n += week - prev[1]
|
||||
elif year == prev[0] + 1 and prev[1] >= 52 and week <= 1:
|
||||
n += 1
|
||||
else:
|
||||
n += max(2, (year - prev[0]) * 53 + week - prev[1])
|
||||
out[(year, week)] = n
|
||||
prev = (year, week)
|
||||
return out
|
||||
|
||||
|
||||
def short_days(item: Period, grain: str) -> int | None:
|
||||
"""这一期不到整期（按年截断的周号算法里跨年切开的那两期、WW 的最后一周）时有几天；整期、不知道起止的为 None。"""
|
||||
if grain != "week" or item.start is None or item.end is None:
|
||||
return None
|
||||
days = (item.end - item.start).days + 1
|
||||
return days if days < 7 else None
|
||||
|
||||
|
||||
@lru_cache(maxsize=256)
|
||||
def week_scheme(sql: str, column: str, dialect: str | None) -> str | None:
|
||||
"""期次列 column 是按哪种周号算法写成「YYYY-Www」的（WEEK_SCHEMES 之一）：看最外层 SELECT 里这一列怎么写（从 CTE、子查询
|
||||
里原样取出的往里找），格式化函数（SQLite 的 strftime、MySQL 的 DATE_FORMAT、PostgreSQL / Oracle 的 to_char）的格式串里
|
||||
年份和周号是不是配套的一对（_WEEK_FORMATS），而且只引用了一列。认不出（方言不知道、不是这几个函数、年份和周号不配套或
|
||||
不止一个、只是取了一列现成的字符串）为 None。"""
|
||||
from sqlglot import exp
|
||||
|
||||
pairs = _WEEK_FORMATS.get(dialect or "")
|
||||
if pairs is None or not sql:
|
||||
return None
|
||||
try:
|
||||
value = _period_expr(sql, column, dialect)
|
||||
if value is None:
|
||||
return None
|
||||
refs = {(name_key(c.table), name_key(c.name)) for c in value.find_all(exp.Column)
|
||||
if not isinstance(c.this, exp.Star)}
|
||||
texts = [_native_format(node, dialect) for node in value.find_all(exp.TimeToStr, exp.ToChar)] # type: ignore[arg-type]
|
||||
if len(refs) != 1 or not texts or any(t is None for t in texts):
|
||||
return None
|
||||
marks = [m for t in texts for m in _format_marks(t, dialect)] # type: ignore[arg-type]
|
||||
years = [m for m in marks if m in _YEAR_MARKS[dialect]] # type: ignore[index]
|
||||
weeks = [m for m in marks if m in _WEEK_MARKS[dialect]] # type: ignore[index]
|
||||
if len(years) != 1 or len(weeks) != 1:
|
||||
return None
|
||||
return pairs.get((years[0], weeks[0]))
|
||||
except Exception: # noqa: BLE001 - 认不出就当不知道算法
|
||||
return None
|
||||
|
||||
|
||||
def _period_expr(sql: str, column: str, dialect: str | None) -> Any:
|
||||
"""结果列 column 的写法（sqlglot 节点）：最外层 SELECT 里那一项；是从 CTE、子查询里原样取出的一列时往里找。"""
|
||||
from sqlglot import exp
|
||||
|
||||
from app.engine import cell_format as cf
|
||||
|
||||
tree = coverage.parse_sql(sql, dialect)
|
||||
if not isinstance(tree, exp.Select):
|
||||
return None
|
||||
ctes = {name_key(c.alias_or_name): c.this for c in tree.find_all(exp.CTE)}
|
||||
select, name = tree, column
|
||||
for _ in range(_NEST_MAX + 1):
|
||||
item = cf._projection(select, name)
|
||||
if item is None:
|
||||
return None
|
||||
value = cf._unwrap(item.unalias())
|
||||
if not isinstance(value, exp.Column) or isinstance(value.this, exp.Star):
|
||||
return value
|
||||
inner = _inner_select(select, value, ctes)
|
||||
if inner is None:
|
||||
return value
|
||||
select, name = inner, value.name
|
||||
return None
|
||||
|
||||
|
||||
def _inner_select(select: Any, column: Any, ctes: dict[str, Any]) -> Any:
|
||||
"""外层取的这一列来自哪个 CTE、子查询（那一层的 SELECT）；来自实表、认不准是哪个来源的为 None。"""
|
||||
from sqlglot import exp
|
||||
|
||||
sources = [select.args.get("from_") or select.args.get("from"), *(select.args.get("joins") or [])]
|
||||
found = [s.this for s in sources if s is not None and getattr(s, "this", None) is not None]
|
||||
if column.table:
|
||||
found = [n for n in found if name_key(n.alias_or_name) == name_key(column.table)]
|
||||
if len(found) != 1:
|
||||
return None
|
||||
node = found[0]
|
||||
if isinstance(node, exp.Subquery):
|
||||
inner = node.this
|
||||
elif isinstance(node, exp.Table) and not node.db and name_key(node.name) in ctes:
|
||||
inner = ctes[name_key(node.name)]
|
||||
else:
|
||||
return None
|
||||
return inner if isinstance(inner, exp.Select) else None
|
||||
|
||||
|
||||
def _native_format(node: Any, dialect: str) -> str | None:
|
||||
"""格式化函数的格式串，按这个方言原本的写法（sqlglot 解析时换成了自己的写法，再换回去；Oracle 的 to_char 原样留着）。"""
|
||||
from sqlglot import exp
|
||||
from sqlglot.dialects.dialect import Dialect
|
||||
|
||||
fmt = node.args.get("format")
|
||||
if not isinstance(fmt, exp.Literal) or not fmt.is_string:
|
||||
return None
|
||||
if isinstance(node, exp.ToChar):
|
||||
return str(fmt.this)
|
||||
text = Dialect.get_or_raise(dialect).generator().format_time(node)
|
||||
if not text or len(text) < 2 or text[0] != "'" or text[-1] != "'":
|
||||
return None
|
||||
return text[1:-1].replace("''", "'")
|
||||
|
||||
|
||||
def _format_marks(text: str, dialect: str) -> list[str]:
|
||||
"""格式串里的格式符（按先后）：strftime、DATE_FORMAT 是「%」加一个字（%% 是百分号本身）；to_char 去掉双引号里的原样文字后
|
||||
按 _CHAR_MARKS 从左往右认（不分大小写）。"""
|
||||
out: list[str] = []
|
||||
i = 0
|
||||
if dialect in ("sqlite", "mysql"):
|
||||
while i < len(text):
|
||||
if text[i] == "%" and i + 1 < len(text):
|
||||
if text[i + 1] != "%":
|
||||
out.append(text[i:i + 2])
|
||||
i += 2
|
||||
else:
|
||||
i += 1
|
||||
return out
|
||||
plain = re.sub(r'"[^"]*"', "", text).upper()
|
||||
while i < len(plain):
|
||||
mark = next((m for m in _CHAR_MARKS if plain.startswith(m, i)), None)
|
||||
if mark is not None:
|
||||
out.append(mark)
|
||||
i += len(mark)
|
||||
else:
|
||||
i += 1
|
||||
return out
|
||||
|
||||
|
||||
def _infer_date_grain(days: Sequence[date]) -> str:
|
||||
"""一列日期的粒度（见文件头）。"""
|
||||
ordered = sorted(set(days))
|
||||
@@ -131,8 +376,10 @@ def _infer_date_grain(days: Sequence[date]) -> str:
|
||||
return "day"
|
||||
|
||||
|
||||
def parse(values: Sequence[Any], grain: str | None = None) -> tuple[list[Period], str] | None:
|
||||
"""一列期次值 → (逐个的 Period，粒度)。认不出（写法不一、带时刻、粒度和写法对不上）返回 None。"""
|
||||
def parse(values: Sequence[Any], grain: str | None = None, *,
|
||||
weeks: str | None = None) -> tuple[list[Period], str] | None:
|
||||
"""一列期次值 → (逐个的 Period，粒度)。认不出（写法不一、带时刻、粒度和写法对不上、周号在 weeks 这种算法里没有）返回 None，
|
||||
是哪几个值见 unreadable。weeks 是 week_scheme 认出的周号算法：只用于「YYYY-Www」，None 是认不出算法（起止为空）。"""
|
||||
shapes = {_shape(v) for v in values}
|
||||
if not values or None in shapes or len(shapes) != 1:
|
||||
return None
|
||||
@@ -152,16 +399,16 @@ def parse(values: Sequence[Any], grain: str | None = None) -> tuple[list[Period]
|
||||
if shape == "week":
|
||||
if grain not in (None, "week"):
|
||||
return None
|
||||
out = []
|
||||
for v in values:
|
||||
m = _WEEK.match(str(v).strip())
|
||||
assert m is not None
|
||||
try:
|
||||
start = date.fromisocalendar(int(m.group(1)), int(m.group(2)), 1)
|
||||
except ValueError:
|
||||
return None
|
||||
out.append(Period(v, start.toordinal() // 7, start, start + timedelta(days=6)))
|
||||
return out, "week"
|
||||
labels = [_week_label(v) for v in values]
|
||||
if weeks is None:
|
||||
order = _loose_ordinals(labels)
|
||||
return [Period(v, order[label], None, None) for v, label in zip(values, labels)], "week"
|
||||
spans = [week_span(weeks, *label) for label in labels]
|
||||
if any(s is None for s in spans):
|
||||
return None
|
||||
base = min(year for year, _ in labels)
|
||||
return [Period(v, _week_ordinal(weeks, s, label[0], base), s[0], s[1]) # type: ignore[index, arg-type]
|
||||
for v, label, s in zip(values, labels, spans)], "week"
|
||||
if shape == "quarter":
|
||||
if grain not in (None, "quarter"):
|
||||
return None
|
||||
@@ -196,6 +443,33 @@ def parse(values: Sequence[Any], grain: str | None = None) -> tuple[list[Period]
|
||||
return None
|
||||
|
||||
|
||||
def unreadable(values: Sequence[Any], grain: str | None = None, *,
|
||||
weeks: str | None = None) -> tuple[str, list[Any]]:
|
||||
"""parse 认不出时，为什么、是哪几个值（按先后、去重）：shape 写法认不出（空值、带时刻、不是这几种写法的）；mixed 写法混着
|
||||
（不是最多那种写法的几个，一样多时以先出现的为准）；week 周号在 SQL 的算法里没有这一周；grain 和给的粒度对不上（全部）。"""
|
||||
shapes = [_shape(v) for v in values]
|
||||
odd = [v for v, s in zip(values, shapes) if s is None]
|
||||
kinds = list(dict.fromkeys(shapes))
|
||||
if odd or not values:
|
||||
return "shape", _distinct(odd)
|
||||
if len(kinds) > 1:
|
||||
main = max(kinds, key=shapes.count)
|
||||
return "mixed", _distinct(v for v, s in zip(values, shapes) if s != main)
|
||||
if kinds[0] == "week" and weeks is not None and grain in (None, "week"):
|
||||
odd = [v for v in values if week_span(weeks, *_week_label(v)) is None]
|
||||
if odd:
|
||||
return "week", _distinct(odd)
|
||||
return "grain", _distinct(values)
|
||||
|
||||
|
||||
def _distinct(values: Iterable[Any]) -> list[Any]:
|
||||
out: list[Any] = []
|
||||
for v in values:
|
||||
if not any(v == o and type(v) is type(o) for o in out):
|
||||
out.append(v)
|
||||
return out
|
||||
|
||||
|
||||
def calendar_unit(start: date, end: date) -> str | None:
|
||||
"""这段日子是不是一个整的日历期：整月 month、整季 quarter、整年 year；都不是为 None。整月对整月算一样长（31 天的 10 月
|
||||
和 30 天的 9 月环比是通行的比法），不按天数判窗口不等。"""
|
||||
@@ -221,4 +495,5 @@ def weekend_days(start: date, end: date) -> int:
|
||||
return count
|
||||
|
||||
|
||||
__all__ = ["COUNT_WORD", "DEFAULT_LAG", "GRAINS", "PER_WORD", "Period", "calendar_unit", "parse", "weekend_days"]
|
||||
__all__ = ["COUNT_WORD", "DEFAULT_LAG", "GRAINS", "PER_WORD", "WEEK_SCHEMES", "Period", "calendar_unit", "parse",
|
||||
"short_days", "unreadable", "week_scheme", "week_span", "weekend_days"]
|
||||
|
||||
@@ -29,7 +29,7 @@ from typing import Any
|
||||
from app.analysis.contract import OP_LABELS
|
||||
from app.analysis.numbers import fixed, to_fraction
|
||||
|
||||
#: 附带说明进判定语的先后；不在这里的（measure_unlabeled、headline_basis）只在证据面板
|
||||
#: 附带说明进判定语的先后；不在这里的（measure_unlabeled、headline_basis、partial_periods）只在证据面板
|
||||
PHRASE_ORDER = ("comparability", "simpson", "offset", "outlier_sensitive", "small_sample", "record_high", "record_low",
|
||||
"multiple_comparisons", "weekday_effect", "small_denominator", "null_values")
|
||||
#: 判定语里最多接几条附带说明
|
||||
@@ -306,6 +306,13 @@ def small_sample_mean_text(group: str, n: int, minimum: int) -> str:
|
||||
|
||||
|
||||
UNVERIFIED_COMPLETE = "没能核对本期的数据是否到齐（查不到表里这一列的日期范围）"
|
||||
UNKNOWN_WEEKS = "没能核对本期的数据是否到齐（期次写成「年-W周号」，认不出 SQL 是按哪种周号算的，不知道是哪几天）"
|
||||
NO_DATES = "没能核对本期的数据是否到齐（期次不是日期，不知道是哪几天）"
|
||||
|
||||
|
||||
def partial_periods_text(count: int) -> str:
|
||||
"""按年截断的周号算法里跨年切开的那几期（不到整周）没有算进历史。"""
|
||||
return f"有 {count} 期不到整周（跨年被切开），没有算进历史"
|
||||
|
||||
|
||||
def with_caveats(main: str, caveats: Sequence[Mapping[str, str]], *, skip: Iterable[str] = ()) -> str:
|
||||
@@ -501,14 +508,15 @@ CRITERION_COMPARABLE = ("只差日期、窗口一样长、是整期、星期构
|
||||
|
||||
|
||||
__all__ = [
|
||||
"CRITERION_COMPARABLE", "EMPTY_KEY", "HEADLINE_BASIS", "KEY_JOIN", "LABEL_MAX", "MEASURE_WORDS",
|
||||
"PHRASE_CAVEATS", "PHRASE_ORDER", "UNVERIFIED_COMPLETE", "WEEKDAY_EFFECT", "amount", "band_text",
|
||||
"CRITERION_COMPARABLE", "EMPTY_KEY", "HEADLINE_BASIS", "KEY_JOIN", "LABEL_MAX", "MEASURE_WORDS", "NO_DATES",
|
||||
"PHRASE_CAVEATS", "PHRASE_ORDER", "UNKNOWN_WEEKS", "UNVERIFIED_COMPLETE", "WEEKDAY_EFFECT", "amount", "band_text",
|
||||
"baseline_flat_phrase", "baseline_phrase", "caveat", "change", "clean_label", "comparable_phrase",
|
||||
"criterion_additive", "criterion_baseline", "criterion_differ", "criterion_factor", "criterion_ratio",
|
||||
"criterion_trend", "current_text", "decompose_concentrated", "decompose_dispersed", "decompose_driver",
|
||||
"decompose_flat", "decompose_unclosed", "differ_pair", "differ_phrase", "dims_text", "factor_both",
|
||||
"factor_driver", "key_text", "multiple_text", "names", "no_evidence_phrase", "null_periods_text",
|
||||
"null_segments_text", "number", "offset_text", "outlier_text", "pct", "pick_label", "points", "ratio_both",
|
||||
"null_segments_text", "number", "offset_text", "outlier_text", "partial_periods_text", "pct", "pick_label",
|
||||
"points", "ratio_both",
|
||||
"ratio_change", "ratio_mix", "ratio_rate", "ratio_value", "record_text", "refused", "say", "share_words",
|
||||
"simpson_text", "small_denominator_text", "small_sample_mean_text", "small_sample_rate_text", "threshold_pct",
|
||||
"trend_phrase", "unconfirmed_text", "unlabeled_text", "with_caveats",
|
||||
|
||||
@@ -5,9 +5,11 @@ week / month / quarter，缺省按期次值推断，见 analysis/periods）、la
|
||||
（change 缺省：和比较期比的变化率；level：值本身，给存量、比率用）、lag（和往前第几期比，缺省日数据 7、其余 1；同比写 52 /
|
||||
12 / 4，历史要相应够长）。
|
||||
|
||||
**前提**（不满足就算不了，原因照实写）：期次认得出（period_unreadable）、本期完整（同可比性检查的整期：没过完、数据没到、
|
||||
SQL 截掉一部分 → partial_target；表的日期范围查不到时只按运行当天判，附一句核对不了）、本期和比较期之间不缺期（target_gap）、
|
||||
比较期不为 0（zero_base）、历史够长（history_short）；趋势要最后几期连着（中间缺期 series_gap）。度量格为空的那一期跳过。
|
||||
**前提**（不满足就算不了，原因照实写）：期次认得出（period_unreadable，原因里举认不出的那几个值）、本期完整（同可比性检查的
|
||||
整期：不到整周、没过完、数据没到、SQL 截掉一部分 → partial_target；表的日期范围查不到时只按运行当天判，附一句核对不了；期次
|
||||
没有日期、周号认不出算法的不判齐，附一句核对不了）、比较期有值（缺期、那一期的值为空、不到整周 → target_gap，原因照实分开
|
||||
写）、比较期不为 0（zero_base）、历史够长（history_short）；趋势要最后几期连着（中间缺期、值为空、不到整周 series_gap）。度量
|
||||
格为空的那一期跳过；不到整周的那几期（按年截断的周号算法里跨年切开的）不算进历史。
|
||||
|
||||
**波动基线**：历史的中位数 m、中位绝对偏差 MAD，σ̂ = 1.4826·MAD，稳健 z = (x − m)/σ̂；常规波动范围 m ± k_warn·σ̂；全程
|
||||
精确的有理数。σ̂ = 0（历史每期都一样）时本期一样判常规、不一样判明显超出。本期的值（不论 basis）是历史各期里最高 / 最低的
|
||||
@@ -50,6 +52,20 @@ class Series:
|
||||
points: list[tuple[pd.Period, Fraction]] # 按期次排好、度量不为空的
|
||||
nulls: int
|
||||
params: dict[str, Any]
|
||||
periods: dict[int, pd.Period] # 每一期（含度量为空的），按 ordinal
|
||||
empty: set[int] # 度量为空的那几期的 ordinal
|
||||
|
||||
def whole(self, item: pd.Period) -> bool:
|
||||
return pd.short_days(item, self.grain) is None
|
||||
|
||||
def missing(self, ordinal: int) -> tuple[str, str] | None:
|
||||
"""某一期拿不到能用的值的原因：(empty 值为空 / short 不到整周, 那一期的写法)；真缺了这一期为 None。"""
|
||||
item = self.periods.get(ordinal)
|
||||
if item is None:
|
||||
return None
|
||||
if ordinal in self.empty:
|
||||
return "empty", ph.key_text([item.raw])
|
||||
return ("short", ph.key_text([item.raw])) if not self.whole(item) else None
|
||||
|
||||
|
||||
def _load(raw: Mapping[str, Any], env: Env, used: list[str]) -> Series | Any:
|
||||
@@ -71,29 +87,45 @@ def _load(raw: Mapping[str, Any], env: Env, used: list[str]) -> Series | Any:
|
||||
criterion=criterion)
|
||||
i_period = snap.index(period)
|
||||
values = [row[i_period] if i_period < len(row) else None for row in snap.rows]
|
||||
parsed = pd.parse(values, grain)
|
||||
weeks = pd.week_scheme(snap.sql, period, snap.dialect)
|
||||
parsed = pd.parse(values, grain, weeks=weeks)
|
||||
if parsed is None:
|
||||
sample = "、".join(dict.fromkeys(ph.key_text([v]) for v in values[:3]))
|
||||
return refuse(env, params, used, "period_unreadable",
|
||||
f"期次列「{period}」的值认不出是哪一期（{sample}）：要日期、YYYY-MM、YYYY-Www、YYYY-Qn 或整数序号",
|
||||
return refuse(env, params, used, "period_unreadable", unreadable_text(period, values, grain, weeks),
|
||||
criterion=criterion)
|
||||
items, grain = parsed
|
||||
params["grain"] = grain
|
||||
seen: set[int] = set()
|
||||
periods: dict[int, pd.Period] = {}
|
||||
empty: set[int] = set()
|
||||
points: list[tuple[pd.Period, Fraction]] = []
|
||||
nulls = 0
|
||||
for r, item in enumerate(items):
|
||||
if item.ordinal in seen:
|
||||
if item.ordinal in periods:
|
||||
raise CallError("duplicate_rows", ph.say(f"{snap.ref} 里同一期有不止一行：在 SQL 里按 ",
|
||||
column_label(env, snap, period), " 汇总，每期一行"))
|
||||
seen.add(item.ordinal)
|
||||
periods[item.ordinal] = item
|
||||
value = snap.number(r, measure, "度量列")
|
||||
if value is None:
|
||||
nulls += 1
|
||||
empty.add(item.ordinal)
|
||||
continue
|
||||
points.append((item, value))
|
||||
points.sort(key=lambda p: p[0].ordinal)
|
||||
return Series(snap, period, measure, grain, label, points, nulls, params)
|
||||
return Series(snap, period, measure, grain, label, points, len(empty), params, periods, empty)
|
||||
|
||||
|
||||
#: 给了粒度、期次对不上时的说法
|
||||
_GRAIN_WORD = {"day": "按天", "week": "按周", "month": "按月", "quarter": "按季度", "seq": "按序号"}
|
||||
|
||||
|
||||
def unreadable_text(period: str, values: list[Any], grain: str | None, weeks: str | None) -> str:
|
||||
"""期次认不出的原因：举出认不出的那几个值（最多三个），按为什么认不出分开说（periods.unreadable）。"""
|
||||
kind, odd = pd.unreadable(values, grain, weeks=weeks)
|
||||
shown = "、".join("空值" if v is None else ph.key_text([v]) for v in odd[:3]) + ("等" if len(odd) > 3 else "")
|
||||
if kind == "mixed":
|
||||
return ph.say(f"期次列「{period}」的写法不一：", shown, "和其余的不是同一种写法")
|
||||
if kind == "week":
|
||||
return ph.say(f"期次列「{period}」里的", shown, "按 SQL 里的周号算法没有这一周")
|
||||
if kind == "grain":
|
||||
return ph.say(f"期次列「{period}」的值（", shown, f"）对不上{_GRAIN_WORD.get(str(grain), '给的')}的粒度")
|
||||
return f"期次列「{period}」的值认不出是哪一期（{shown}）：要日期、YYYY-MM、YYYY-Www、YYYY-Qn 或整数序号"
|
||||
|
||||
|
||||
def _criterion(env: Env, n: int | None, unit: str) -> str:
|
||||
@@ -102,13 +134,11 @@ def _criterion(env: Env, n: int | None, unit: str) -> str:
|
||||
return ph.criterion_trend(env.settings, n, unit)
|
||||
|
||||
|
||||
def _completeness(env: Env, series: Series, item: pd.Period) -> tuple[str, str | None]:
|
||||
"""本期齐不齐（同可比性检查的整期）：(pass / short / unknown, 不齐的那一句)。"""
|
||||
def _completeness(env: Env, series: Series, item: pd.Period) -> tuple[str, str | None, str | None]:
|
||||
"""本期齐不齐（同可比性检查的整期，comparable.period_complete）：(pass / short / unknown, 不齐的那一句, 核对不了时的附带说明)。"""
|
||||
series.params["today"] = today_param(env)
|
||||
if item.start is None:
|
||||
return "unknown", None
|
||||
span = cmp.period_span("本期", series.snap, series.period, item, series.grain, series.measure, env.facts, env.book)
|
||||
return cmp.span_coverage(span, env.today, env.book) # type: ignore[arg-type]
|
||||
return cmp.period_complete(series.snap, series.period, item, series.grain, series.measure, env.facts, env.book,
|
||||
env.today) # type: ignore[arg-type]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -144,18 +174,19 @@ def baseline(raw: Mapping[str, Any], env: Env) -> Any:
|
||||
target, x_level = found
|
||||
params["target"] = target.raw
|
||||
caveats: list[dict[str, str]] = []
|
||||
status, why = _completeness(env, series, target)
|
||||
status, why, note = _completeness(env, series, target)
|
||||
if status == "short":
|
||||
return refuse(env, params, used, "partial_target", f"{why}，和完整的{per}比没有意义", criterion=criterion)
|
||||
if status == "unknown" and target.start is not None:
|
||||
caveats.append(ph.caveat("comparability", ph.UNVERIFIED_COMPLETE))
|
||||
by_ordinal = {p.ordinal: v for p, v in series.points}
|
||||
before = [(p, v) for p, v in series.points if p.ordinal < target.ordinal]
|
||||
if status == "unknown" and note:
|
||||
caveats.append(ph.caveat("comparability", note))
|
||||
# 不到整周的那几期（跨年切开的）既不当比较期、也不算进历史
|
||||
by_ordinal = {p.ordinal: v for p, v in series.points if series.whole(p)}
|
||||
before = [(p, v) for p, v in series.points if p.ordinal < target.ordinal and series.whole(p)]
|
||||
cut = sum(1 for p, _ in series.points if p.ordinal < target.ordinal and not series.whole(p))
|
||||
if basis == "change":
|
||||
prev = by_ordinal.get(target.ordinal - lag)
|
||||
if prev is None:
|
||||
return refuse(env, params, used, "target_gap",
|
||||
f"本期和它的比较期（往前第 {lag} {'期' if series.grain == 'seq' else unit}）之间缺期，没有比较期的值",
|
||||
return refuse(env, params, used, "target_gap", _compare_gap(series, target.ordinal - lag, lag, unit),
|
||||
criterion=criterion)
|
||||
if prev == 0:
|
||||
return refuse(env, params, used, "zero_base", ph.say("比较期的", series.label, "为 0，算不了变化率"),
|
||||
@@ -190,6 +221,8 @@ def baseline(raw: Mapping[str, Any], env: Env) -> Any:
|
||||
caveats.append(ph.caveat("record_low", ph.record_text(False, series.label, n, unit)))
|
||||
if series.nulls:
|
||||
caveats.append(ph.caveat("null_values", ph.null_periods_text(series.nulls, series.label)))
|
||||
if cut:
|
||||
caveats.append(ph.caveat("partial_periods", ph.partial_periods_text(cut)))
|
||||
current = ph.current_text(x, basis=basis, lag=lag, grain=series.grain)
|
||||
if sigma == 0:
|
||||
shown = ph.pct(m) if basis == "change" else ph.amount(m)
|
||||
@@ -213,6 +246,17 @@ def baseline(raw: Mapping[str, Any], env: Env) -> Any:
|
||||
phrase=ph.with_caveats(main, caveats, skip=("comparability",)), caveats=caveats)
|
||||
|
||||
|
||||
def _compare_gap(series: Series, ordinal: int, lag: int, unit: str) -> str:
|
||||
"""比较期拿不到值的原因，照实分开说：那一期的值为空、那一期不到整周、真缺了那一期。"""
|
||||
found = series.missing(ordinal)
|
||||
if found is not None and found[0] == "empty":
|
||||
return ph.say("比较期（", found[1], "）的", series.label, "为空，没有比较期的值")
|
||||
if found is not None:
|
||||
days = pd.short_days(series.periods[ordinal], series.grain)
|
||||
return ph.say("比较期（", found[1], f"）只有 {days} 天，不是整周，和它比没有意义")
|
||||
return f"本期和它的比较期（往前第 {lag} {'期' if series.grain == 'seq' else unit}）之间缺期，没有比较期的值"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 趋势
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -245,15 +289,15 @@ def trend(raw: Mapping[str, Any], env: Env) -> Any:
|
||||
return refuse(env, params, used, "history_short", f"只有 0 期，至少要 {minimum} 期", criterion=criterion)
|
||||
last = series.points[-1][0]
|
||||
caveats: list[dict[str, str]] = []
|
||||
status, why = _completeness(env, series, last)
|
||||
status, why, note = _completeness(env, series, last)
|
||||
if status == "short":
|
||||
return refuse(env, params, used, "partial_target", f"{why}，和完整的{per}比没有意义", criterion=criterion)
|
||||
if status == "unknown" and last.start is not None:
|
||||
caveats.append(ph.caveat("comparability", ph.UNVERIFIED_COMPLETE))
|
||||
if status == "unknown" and note:
|
||||
caveats.append(ph.caveat("comparability", note))
|
||||
cap = int(env.settings["max_history"]) + 1
|
||||
tail = [series.points[-1]]
|
||||
for point in reversed(series.points[:-1]):
|
||||
if len(tail) >= cap or point[0].ordinal != tail[-1][0].ordinal - 1:
|
||||
if len(tail) >= cap or point[0].ordinal != tail[-1][0].ordinal - 1 or not series.whole(point[0]):
|
||||
break
|
||||
tail.append(point)
|
||||
tail.reverse()
|
||||
@@ -263,7 +307,8 @@ def trend(raw: Mapping[str, Any], env: Env) -> Any:
|
||||
gap = len(series.points) > n and len(tail) < cap
|
||||
if gap:
|
||||
return refuse(env, params, used, "series_gap",
|
||||
f"最后连着的只有 {n} 期，再往前缺期（至少要连着的 {minimum} 期）", criterion=criterion)
|
||||
f"最后连着的只有 {n} 期，{_tail_gap(series, tail[0][0].ordinal - 1)}（至少要连着的 {minimum} 期）",
|
||||
criterion=criterion)
|
||||
return refuse(env, params, used, "history_short", f"只有 {n} 期，至少要 {minimum} 期", criterion=criterion)
|
||||
ys = [v for _, v in tail]
|
||||
xs = [Fraction(i) for i in range(n)]
|
||||
@@ -302,4 +347,14 @@ def trend(raw: Mapping[str, Any], env: Env) -> Any:
|
||||
phrase=ph.with_caveats(main, caveats, skip=("comparability",)), caveats=caveats)
|
||||
|
||||
|
||||
__all__ = ["MAD_SCALE", "baseline", "run_length", "trend"]
|
||||
def _tail_gap(series: Series, ordinal: int) -> str:
|
||||
"""最后连着的一段往前断开的原因：那一期的值为空、不到整周，或者真缺了那一期。"""
|
||||
found = series.missing(ordinal)
|
||||
if found is not None and found[0] == "empty":
|
||||
return ph.say("再往前的", found[1], "那一期的", series.label, "为空")
|
||||
if found is not None:
|
||||
return ph.say("再往前的", found[1], f"只有 {pd.short_days(series.periods[ordinal], series.grain)} 天，不是整周")
|
||||
return "再往前缺期"
|
||||
|
||||
|
||||
__all__ = ["MAD_SCALE", "baseline", "run_length", "trend", "unreadable_text"]
|
||||
|
||||
@@ -181,3 +181,26 @@ async def test_a_masked_date_column_is_neither_used_nor_probed():
|
||||
period("Q2", "2026-10-12", "2026-10-19", masked=["order_date"]), probe=probe, measure="gmv")
|
||||
assert probe.calls == [] and result["probes"] == []
|
||||
assert result["verdict"]["code"] == "comparable" and "窗口长度" in result["phrase"]
|
||||
|
||||
|
||||
async def test_one_snapshot_of_sqlite_weeks_uses_the_real_days_of_each_week():
|
||||
# SQLite 的 strftime('%Y-W%W')：2026-W42 是 10-19 到 10-25（运行当天 10-21 还没过完），跨年的 2025-W52、2026-W00 只有 3 天、4 天
|
||||
sql = "SELECT strftime('%Y-W%W', order_date) AS week, COUNT(*) AS orders FROM orders GROUP BY week"
|
||||
rows = [["2025-W52", 300], ["2026-W00", 400], ["2026-W01", 700], ["2026-W41", 700], ["2026-W42", 200]]
|
||||
s = snap("Q1", sql, ["week", "orders"], rows, types={"week": "text", "orders": "number"})
|
||||
probe = fixed_prober({("orders", "order_date"): ("2025-01-01", "2026-10-20")})
|
||||
now = await analyze("comparable", snaps=[s], probe=probe, query="Q1", period="week", base="2026-W41",
|
||||
target="2026-W42")
|
||||
assert now["verdict"]["code"] == "incomparable"
|
||||
assert now["phrase"] == "两期不可比：本期数据只到 2026-10-20，还差 5 天"
|
||||
assert now["values"]["target_start"] == "2026-10-19" and now["values"]["target_end"] == "2026-10-25"
|
||||
cut = await analyze("comparable", snaps=[s], probe=probe, query="Q1", period="week", base="2025-W52",
|
||||
target="2026-W00")
|
||||
assert cut["verdict"]["code"] == "incomparable"
|
||||
assert "两段的天数不同：3 天对 4 天" in cut["phrase"]
|
||||
assert "本期只有 2026-01-01 到 2026-01-04 这 4 天，不是整周" in comparable_reasons(cut)
|
||||
assert checks(cut)["星期构成"] == "提醒"
|
||||
# 不知道方言（认不出周号算法）：窗口、整期、星期构成都核对不了，不判可比以外的结论
|
||||
plain = snap("Q1", sql, ["week", "orders"], rows, types={"week": "text", "orders": "number"})
|
||||
unknown = await analyze("comparable", snaps=[plain], query="Q1", period="week", base="2026-W41", target="2026-W42")
|
||||
assert unknown["phrase"] == "两期可比，窗口长度、整期、星期构成核对不了"
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""期次（app/analysis/periods.py）：日期、YYYY-MM、YYYY-Www、YYYY-Qn、整数序号认得出、能排序、知道覆盖哪几天；粒度按值推断；
|
||||
写法混着、带时刻的认不出。"""
|
||||
写法混着、带时刻的认不出。周号按 SQL 里的格式串认算法（SQLite 的 %W / %U / %G-%V、MySQL 的 %U / %u / %X-%V / %x-%v、to_char
|
||||
的 WW / IW），按算法算真实起止；认不出算法的起止为空。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
@@ -26,10 +27,10 @@ def test_labels_for_months_weeks_quarters_and_sequences():
|
||||
assert grain == "month" and items[1].ordinal - items[0].ordinal == 1 and items[1].end == date(2026, 10, 31)
|
||||
items, grain = pd.parse([202609, 202610])
|
||||
assert grain == "month"
|
||||
items, grain = pd.parse(["2026-W41", "2026-W42", "2026-W01"])
|
||||
items, grain = pd.parse(["2026-W41", "2026-W42", "2026-W01"], weeks="iso")
|
||||
assert grain == "week" and items[1].start == date(2026, 10, 12) and items[1].end == date(2026, 10, 18)
|
||||
assert items[1].ordinal - items[0].ordinal == 1
|
||||
items, grain = pd.parse(["2025-W52", "2026-W01"]) # 跨年也连着
|
||||
items, grain = pd.parse(["2025-W52", "2026-W01"], weeks="iso") # 跨年也连着
|
||||
assert items[1].ordinal - items[0].ordinal == 1
|
||||
items, grain = pd.parse(["2026Q3", "2026-Q4"])
|
||||
assert grain == "quarter" and items[1].start == date(2026, 10, 1)
|
||||
@@ -55,3 +56,75 @@ def test_calendar_units_and_weekend_days():
|
||||
assert pd.weekend_days(date(2026, 10, 16), date(2026, 10, 18)) == 2 # 周五到周日
|
||||
assert pd.weekend_days(date(2026, 10, 12), date(2026, 10, 14)) == 0
|
||||
assert pd.weekend_days(date(2026, 10, 1), date(2026, 10, 31)) == 9
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 周号：按 SQL 认算法
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
WEEK_SQL = {
|
||||
("sqlite", "SELECT strftime('%Y-W%W', day) AS week, SUM(amount) AS gmv FROM orders GROUP BY week"): "monday",
|
||||
("sqlite", "SELECT strftime('%Y-W%U', day) AS week FROM orders GROUP BY 1"): "sunday",
|
||||
("sqlite", "SELECT strftime('%G-W%V', day) AS week FROM orders GROUP BY 1"): "iso",
|
||||
("sqlite", "SELECT strftime('%Y', day) || '-W' || strftime('%W', day) AS week FROM orders GROUP BY 1"): "monday",
|
||||
("sqlite", "WITH w AS (SELECT strftime('%Y-W%W', day) AS wk, amount FROM orders) "
|
||||
"SELECT wk AS week, SUM(amount) AS gmv FROM w GROUP BY 1"): "monday",
|
||||
("sqlite", "SELECT t.wk AS week FROM (SELECT strftime('%G-W%V', day) AS wk FROM orders) AS t GROUP BY 1"): "iso",
|
||||
("mysql", "SELECT DATE_FORMAT(day, '%x-W%v') AS week FROM orders GROUP BY 1"): "iso",
|
||||
("mysql", "SELECT DATE_FORMAT(day, '%X-W%V') AS week FROM orders GROUP BY 1"): "sunday_year",
|
||||
("mysql", "SELECT DATE_FORMAT(day, '%Y-W%u') AS week FROM orders GROUP BY 1"): "monday4",
|
||||
("mysql", "SELECT DATE_FORMAT(day, '%Y-W%U') AS week FROM orders GROUP BY 1"): "sunday",
|
||||
("postgres", "SELECT to_char(day, 'IYYY-\"W\"IW') AS week FROM orders GROUP BY 1"): "iso",
|
||||
("postgres", "SELECT to_char(day, 'YYYY-\"W\"WW') AS week FROM orders GROUP BY 1"): "jan1",
|
||||
("oracle", "SELECT TO_CHAR(day, 'IYYY-\"W\"IW') AS week FROM orders GROUP BY TO_CHAR(day, 'IYYY-\"W\"IW')"): "iso",
|
||||
("oracle", "SELECT to_char(day, 'YYYY\"W\"WW') AS week FROM orders"): "jan1",
|
||||
}
|
||||
|
||||
|
||||
def test_the_week_algorithm_is_read_from_the_sql():
|
||||
for (dialect, sql), scheme in WEEK_SQL.items():
|
||||
assert pd.week_scheme(sql, "week", dialect) == scheme, (dialect, sql)
|
||||
|
||||
|
||||
def test_week_algorithms_that_cannot_be_told_are_unknown():
|
||||
# 年份和周号不配套：同一个标签会落在两段日子上
|
||||
assert pd.week_scheme("SELECT strftime('%Y-W%V', day) AS week FROM orders", "week", "sqlite") is None
|
||||
assert pd.week_scheme("SELECT DATE_FORMAT(day, '%Y-W%v') AS week FROM orders", "week", "mysql") is None
|
||||
assert pd.week_scheme("SELECT to_char(day, 'YYYY-\"W\"IW') AS week FROM orders", "week", "postgres") is None
|
||||
# MySQL 的 %W 是星期几的名字，不是周号；方言不知道；取的是现成的一列；两个日期列
|
||||
assert pd.week_scheme("SELECT DATE_FORMAT(day, '%Y-W%W') AS week FROM orders", "week", "mysql") is None
|
||||
assert pd.week_scheme("SELECT strftime('%Y-W%W', day) AS week FROM orders", "week", None) is None
|
||||
assert pd.week_scheme("SELECT week, SUM(amount) AS gmv FROM weekly GROUP BY week", "week", "sqlite") is None
|
||||
assert pd.week_scheme("SELECT strftime('%Y', a) || '-W' || strftime('%W', b) AS week FROM t", "week", "sqlite") \
|
||||
is None
|
||||
assert pd.week_scheme("SELECT strftime('%Y-W%W', day) AS wk FROM orders", "week", "sqlite") is None
|
||||
|
||||
|
||||
def test_sqlite_monday_weeks_cut_at_new_year():
|
||||
items, grain = pd.parse(["2025-W52", "2026-W00", "2026-W01", "2026-W42"], weeks="monday")
|
||||
spans = [(i.start, i.end) for i in items]
|
||||
assert spans == [(date(2025, 12, 29), date(2025, 12, 31)), (date(2026, 1, 1), date(2026, 1, 4)),
|
||||
(date(2026, 1, 5), date(2026, 1, 11)), (date(2026, 10, 19), date(2026, 10, 25))]
|
||||
assert [i.ordinal - items[0].ordinal for i in items[:3]] == [0, 1, 2] # 跨年切开的两期也连着
|
||||
assert [pd.short_days(i, grain) for i in items] == [3, 4, None, None]
|
||||
# 1 月 1 日就是周一的年份没有第 0 周：认不出，原因里举的是它
|
||||
assert pd.parse(["2024-W00", "2024-W01"], weeks="monday") is None
|
||||
assert pd.unreadable(["2024-W00", "2024-W01"], weeks="monday") == ("week", ["2024-W00"])
|
||||
# ISO 只有 52 周的年份没有第 53 周
|
||||
assert pd.parse(["2026-W53"], weeks="iso") is not None and pd.parse(["2025-W53"], weeks="iso") is None
|
||||
|
||||
|
||||
def test_weeks_of_an_unknown_algorithm_have_no_dates_but_keep_their_order():
|
||||
items, grain = pd.parse(["2025-W51", "2025-W52", "2026-W00", "2026-W01", "2026-W03"])
|
||||
assert grain == "week" and all(i.start is None and i.end is None for i in items)
|
||||
assert [i.ordinal - items[0].ordinal for i in items] == [0, 1, 2, 3, 5] # 同一年里隔了一周就是缺期
|
||||
items, _ = pd.parse(["2025-W52", "2026-W01"]) # 上一年最后一周接下一年第一周
|
||||
assert items[1].ordinal - items[0].ordinal == 1
|
||||
items, _ = pd.parse(["2025-W40", "2026-W01"])
|
||||
assert items[1].ordinal - items[0].ordinal > 1
|
||||
|
||||
|
||||
def test_unreadable_values_are_named():
|
||||
assert pd.unreadable(["2026-W40", "上周", "2026-W41", None]) == ("shape", ["上周", None])
|
||||
assert pd.unreadable(["2026-W40", "2026-W41", "2026-10-12"]) == ("mixed", ["2026-10-12"])
|
||||
assert pd.unreadable(["2026-10", "2026-11"], "week") == ("grain", ["2026-10", "2026-11"])
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
lag 7、新高新低；上升 / 下降 / 无趋势、连续期数、缺期。每个用例都顺带核对复算和运行逐字段相同。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
from datetime import date, timedelta
|
||||
from fractions import Fraction
|
||||
|
||||
@@ -210,3 +211,105 @@ async def test_daily_trends_warn_about_weekday_effects():
|
||||
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"]
|
||||
|
||||
Reference in New Issue
Block a user