From 7f949cce9243e4f0b73c29a248661972ec8bbc6a Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 17 Oct 2026 07:59:19 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=96=E5=80=BC=E8=A7=82=E5=AF=9F=EF=BC=9A?= =?UTF-8?q?=E4=B8=9A=E5=8A=A1=E6=97=A5=E6=9C=9F=E6=9C=89=E6=9C=AA=E6=9D=A5?= =?UTF-8?q?=E5=80=BC=E6=97=B6=E4=B8=8D=E5=86=8D=E8=AF=AF=E6=8A=A5=E9=87=8F?= =?UTF-8?q?=E8=B7=8C=E3=80=81=E7=BC=BA=E5=A4=A9=EF=BC=9B=E3=80=8C=E4=BB=8A?= =?UTF-8?q?=E5=A4=A9=E3=80=8D=E7=BB=9F=E4=B8=80=E6=8C=89=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E7=AB=AF=E6=89=80=E5=9C=A8=E6=97=B6=E5=8C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 验收复现：业务日期最晚到明天、今天只写了几行，记出一条 volume_drop。原因有两层： - 比较器 value_drift.daily() 取 last = max(days)，只在 last >= today 时退一天；max 落在明天时退到的是今天这个 还没写完的日子。改成最近一个整天 = min(有数据的最后一天, 昨天)。 - 按日行数的查询 day_volume_sql 只有下界，未来的日子也进了窗口。加上界：数到观察当天（含），第二天起不算 （< 第二天，时间列上当天的时分秒照样算进来，文字存的日期按字符串比同样成立）。签名改成必须传 until。 缺天同样的问题也修了：今天以后的空日子不算缺天；以前那次观察只看到它自己那天（data.until）为止，免得它 窗口里未来的空日子被当成「以前就有过缺天」，挡住这一次真缺的天。断档遇到未来值时离今天是负数，只会报不出、 不会误报，观察本身照实记最新业务日期（目录里也照实给人看），不截未来值，写进了模块说明。 「今天」：原来窗口用 UTC 日期、比较器用 now.date()（也是 UTC），业务日期是本地日期，东八区凌晨 0–8 点会差 一天。统一成服务端所在时区（catalog_profile_observe.local_today，和待办汇总 notify.digest_at 同一个口径）， 写进模块说明。 测试：test_value_drift 两条（未来值不算最近整天、未来空日子不算缺天且历史按 until 截）；test_catalog_observe 两条（SQLite 真跑剖析任务，明天的行不进窗口、不记量跌；local_today 按 TZ 取本地日期）。去掉修复回跑都挂。 金样（tests/fixtures/dialect/*.sql）的 day_volume 一行随上界更新；scripts/dialect_it.py 加了一条「只数到观察 当天」的真库核对。 --- backend/app/data/catalog_profile.py | 22 ++++++---- backend/app/data/catalog_profile_observe.py | 22 ++++++---- backend/app/data/value_drift.py | 26 +++++++---- backend/tests/fixtures/dialect/mariadb.sql | 2 +- backend/tests/fixtures/dialect/mysql.sql | 2 +- backend/tests/fixtures/dialect/oracle.sql | 2 +- backend/tests/fixtures/dialect/postgresql.sql | 2 +- backend/tests/fixtures/dialect/sqlite.sql | 2 +- backend/tests/test_catalog_observe.py | 43 +++++++++++++++++++ backend/tests/test_catalog_profile_units.py | 2 +- backend/tests/test_profile_dialect_golden.py | 2 +- backend/tests/test_value_drift.py | 27 ++++++++++++ scripts/dialect_it.py | 11 ++++- 13 files changed, 133 insertions(+), 32 deletions(-) diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index ed63b2ed..7ea86279 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -45,7 +45,7 @@ import re import time from collections.abc import Awaitable, Callable from dataclasses import asdict, dataclass, field -from datetime import date as _date, datetime, timezone +from datetime import date as _date, datetime, timedelta, timezone from decimal import Decimal from typing import Any, NamedTuple @@ -493,17 +493,23 @@ class SqlDialect: hi = self.date_literal(VALID_DATE_BEFORE, text=text) return f"SELECT MAX({column}) FROM {table} WHERE {column} >= {lo} AND {column} < {hi}" - def _window(self, column: str, since: Any, text: bool) -> str: - """时间窗口的条件：since 起（含），并且早于占位日期 9999-12-31——PostgreSQL 的 infinity、Oracle 的 9999-12-31 - 都大于任何一天，不截掉上端就会在「最近 35 天」里多出一天。""" + def _window(self, column: str, since: Any, text: bool, before: Any = None) -> str: + """时间窗口的条件：since 起（含），并且早于 before（不含；缺省是占位日期 9999-12-31——PostgreSQL 的 infinity、 + Oracle 的 9999-12-31 都大于任何一天，不截掉上端就会在「最近 35 天」里多出一天）。""" return (f"{column} >= {self.date_literal(since, text=text)} " - f"AND {column} < {self.date_literal(VALID_DATE_BEFORE, text=text)}") + f"AND {column} < {self.date_literal(before or VALID_DATE_BEFORE, text=text)}") - def day_volume_sql(self, table: str, column: str, since: Any, *, text: bool = False) -> str: - """最近若干天按天计数：since 起（含）每天几行。只在日期列打头有索引时对大表发（走索引的范围扫描）。""" + def day_volume_sql(self, table: str, column: str, since: Any, until: Any, *, text: bool = False) -> str: + """最近若干天按天计数：since 到 until（都含；until 是观察当天）每天几行。只在日期列打头有索引时对大表发（走索引的 + 范围扫描）。上界是 until 的第二天（不含）：业务日期有未来值（预约、计划类的日期，或者写错的）时，今天以后的日子 + 不进窗口——它们不是「最近的整天」，也不该被当成缺天（data/value_drift）。时间列上「< 第二天」把 until 当天的 + 时分秒都算进来；文字存的日期按字符串比，同样成立。""" + last = until if isinstance(until, _date) else _date.fromisoformat(str(until)) + if isinstance(last, datetime): + last = last.date() day = self.day_trunc(column) return (f"SELECT {day} AS d, COUNT(*) AS n FROM {table} " - f"WHERE {self._window(column, since, text)} GROUP BY {day} ORDER BY d") + f"WHERE {self._window(column, since, text, last + timedelta(days=1))} GROUP BY {day} ORDER BY d") def window_value_counts_sql(self, table: str, column: str, date_column: str, since: Any, n: int, *, text: bool = False) -> str: diff --git a/backend/app/data/catalog_profile_observe.py b/backend/app/data/catalog_profile_observe.py index 78b07a31..0663312e 100644 --- a/backend/app/data/catalog_profile_observe.py +++ b/backend/app/data/catalog_profile_observe.py @@ -8,7 +8,11 @@ - rows：这张表的行数（剖析开头估行数那一条的结果，不另发查询）；口径 stats（数据库统计的估算值）/ count（数到上限）； - codes_observe：码值列的取值分布——**所有**码值列都观察，包括人工确认过、标了已列全的（目录-D6）；只读，不改含义； - freshness：业务日期列（已确认或已验证）的最新日期； -- daily_volume：业务日期最近 35 天每天的行数。 +- daily_volume：业务日期最近 35 天每天的行数，数到观察当天为止（今天以后的日子不进窗口，见 SqlDialect.day_volume_sql）。 + +**「今天」按服务端所在时区**（local_today，和待办汇总的 notify.digest_at 同一个口径）：窗口从哪天起、数到哪天、比较器的 +「今天」（最近一个整天是哪天、断档离今天几天）都用它。业务日期是本地日期，按 UTC 的日期算，东八区凌晨 0–8 点的「今天」 +还是前一天，这几样都会差一天。服务端和业务库不在一个时区时，差的那几个小时照实差着（业务库的时区探查不到）。 **大表的口径**（目录-D7）：小表（行数不超过整表统计的上限）看整表（full）。大表：业务日期已确认或已验证、而且这一列打头 有索引（探查记下的 index_first_columns）时看最近 35 天的窗口（window，走索引的范围扫描）；PostgreSQL / Oracle 没有就块采样 @@ -139,12 +143,13 @@ def sample_percent(rows: int | None, target: int) -> float | None: return next((p for p in SAMPLE_LADDER if p <= want), SAMPLE_LADDER[-1]) -def _today() -> date: - return datetime.now(timezone.utc).date() +def local_today(now: datetime | None = None) -> date: + """观察用的「今天」：服务端所在时区的日期（和待办汇总的 notify.digest_at 同一个口径，见模块说明）。now 缺省是现在。""" + return (now or utcnow()).astimezone().date() def _since() -> date: - return _today() - timedelta(days=WINDOW_DAYS) + return local_today() - timedelta(days=WINDOW_DAYS) def _window_ok(cx: Any, meta: Mapping[str, Any], day: str | None) -> bool: @@ -271,8 +276,9 @@ async def _dates(cx: Any, table: str, kind: str, column: str, meta: Mapping[str, result.observations.append(_observation("freshness", column, {"method": "full"}, {"max": str(value)[:19]}, [])) return - since = _since() - res = await cx.run(d.day_volume_sql(t, c, since, text=text), max_rows=WINDOW_DAYS + 2) + today = local_today() + since = today - timedelta(days=WINDOW_DAYS) + res = await cx.run(d.day_volume_sql(t, c, since, today, text=text), max_rows=WINDOW_DAYS + 2) days: dict[str, int] = {} for row in res.rows: day = str(row[0])[:10] if row and row[0] is not None else "" @@ -280,7 +286,7 @@ async def _dates(cx: Any, table: str, kind: str, column: str, meta: Mapping[str, days[day] = days.get(day, 0) + (cp._as_int(row[1]) or 0) basis = {"method": "window", "column": column, "days": WINDOW_DAYS, "since": since.isoformat()} result.observations.append(_observation("daily_volume", column, basis, - {"since": since.isoformat(), "until": _today().isoformat(), + {"since": since.isoformat(), "until": today.isoformat(), "days": days}, [])) @@ -396,7 +402,7 @@ async def _save_one(session: AsyncSession, source: Any, table: str, obs: Mapping current = vd.Obs(kind, obs["basis_key"], obs["data"], now) past = [vd.Obs(h.kind, h.basis_key, h.data or {}, h.at) for h in history] fresh = 0 - for f in vd.compare(current, past, today=now.date(), known=known, complete=catalog.codes_complete(item)): + for f in vd.compare(current, past, today=local_today(now), known=known, complete=catalog.codes_complete(item)): after = dict(f.after) if isinstance(f.after, Mapping) else {"value": f.after} if f.revoke_complete: version = await _revoke_complete(session, source.id, table, column) diff --git a/backend/app/data/value_drift.py b/backend/app/data/value_drift.py index a165f995..432d947a 100644 --- a/backend/app/data/value_drift.py +++ b/backend/app/data/value_drift.py @@ -13,9 +13,17 @@ 记 info，列出变化最大的几个取值。 - **断档**（date_stalled）：最新业务日期没前进，且这个日期离今天超过 max(2 × 以前前进的中位间隔, 3 天)；要见过至少一次 前进才知道它多久前进一次。 -- **缺天**（date_gap）：窗口里（有数据的第一天到最后一天之间）有日子没数据，而以前同口径的窗口都没有过。 +- **缺天**（date_gap）：窗口里（有数据的第一天到最后一天之间）有日子没数据，而以前同口径的窗口都没有过。只看到观察当天 + 为止：业务日期有未来值（预约、计划类的日期，或者写错的）时，今天以后的空日子不是「没数据」，那几天本来就还没到；以前 + 那次观察也只看到它自己那天（data.until）为止，不然它窗口里未来的空日子会被当成「以前就有过缺天」，挡住这一次真缺的天。 - **量跌**（volume_drop）：最近一个整天的行数不到它前面 28 天中位数的 30%（前 28 天至少 14 天有数据、中位数至少 20 行）。 - 最后一天和观察当天同一天（还在写入）时往前退一天。只看这一次的窗口，不需要历史。 + 最近一个整天是「有数据的最后一天」和「观察当天的前一天」里早的那个：观察当天还在写入，不是整天；业务日期有未来值时 + 有数据的最后一天在今天以后，也不能拿它退一天（那样会退到今天这个不完整的日子上）。只看这一次的窗口，不需要历史。 +- **断档**遇到未来值：最新业务日期在今天以后时离今天是负数，不会报断档（报不出、也不误报）；观察查询本身不截未来值 + （最新业务日期照实记，目录里也照实给人看）。 + +「今天」由调用方传进来（today）：取值观察按服务端所在时区算（data/catalog_profile_observe.local_today），业务日期是本地 +日期，用 UTC 的日期在本地凌晨会差一天。 - **行数**（rows_drop / rows_jump）：比上一次同口径的少了 30% 以上 / 涨到 3 倍以上；依据是数据库统计的估算值时，要连续 两次都这样（和跌、涨之前那一次比）才报。数到上限、只知道「至少多少行」的不比。 """ @@ -177,7 +185,10 @@ def _series(data: Mapping[str, Any]) -> dict[date, int]: return out -def _holes(days: Mapping[date, int]) -> list[date]: +def _holes(days: Mapping[date, int], until: date | None = None) -> list[date]: + """有数据的第一天到最后一天之间没数据的日子；until 给了时只看到这一天为止（今天以后的空日子不算缺天）。""" + if until is not None: + days = {d: n for d, n in days.items() if d <= until} if not days: return [] first, last = min(days), max(days) @@ -190,15 +201,14 @@ def daily(current: Obs, history: Iterable[Obs], *, today: date) -> list[Finding] same, _note = comparable(current, history) days = _series(current.data) out: list[Finding] = [] - holes = _holes(days) - if holes and same and all(not _holes(_series(h.data)) for h in same): + holes = _holes(days, today) + if holes and same and all(not _holes(_series(h.data), _day(h.data.get("until"))) for h in same): out.append(Finding("date_gap", "warning", before={"gaps": 0}, after={"gaps": len(holes), "days": [d.isoformat() for d in holes[:GAPS_SHOWN]]}, key=tuple(d.isoformat() for d in holes))) if days: - last = max(days) - if last >= today: - last -= timedelta(days=1) + # 最近一个整天：有数据的最后一天和昨天里早的那个（今天还在写入；未来的日子不是整天） + last = min(max(days), today - timedelta(days=1)) base = [days.get(last - timedelta(days=i), 0) for i in range(1, VOLUME_BASELINE_DAYS + 1)] present = sum(1 for n in base if n > 0) median = statistics.median(base) if base else 0 diff --git a/backend/tests/fixtures/dialect/mariadb.sql b/backend/tests/fixtures/dialect/mariadb.sql index b837c0b0..75bdc227 100644 --- a/backend/tests/fixtures/dialect/mariadb.sql +++ b/backend/tests/fixtures/dialect/mariadb.sql @@ -22,7 +22,7 @@ SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31 -- date_window SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; -- day_volume -SELECT DATE(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY DATE(paid_at) ORDER BY d; +SELECT DATE(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '2026-10-02' GROUP BY DATE(paid_at) ORDER BY d; -- sample -- （这种方言没有这类查询） -- max_date diff --git a/backend/tests/fixtures/dialect/mysql.sql b/backend/tests/fixtures/dialect/mysql.sql index 17f0f5da..57141703 100644 --- a/backend/tests/fixtures/dialect/mysql.sql +++ b/backend/tests/fixtures/dialect/mysql.sql @@ -22,7 +22,7 @@ SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31 -- date_window SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; -- day_volume -SELECT DATE(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY DATE(paid_at) ORDER BY d; +SELECT DATE(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '2026-10-02' GROUP BY DATE(paid_at) ORDER BY d; -- sample -- （这种方言没有这类查询） -- max_date diff --git a/backend/tests/fixtures/dialect/oracle.sql b/backend/tests/fixtures/dialect/oracle.sql index ba09c3d6..d8c1de40 100644 --- a/backend/tests/fixtures/dialect/oracle.sql +++ b/backend/tests/fixtures/dialect/oracle.sql @@ -22,7 +22,7 @@ SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31 -- date_window SELECT * FROM (SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status) WHERE ROWNUM <= 21; -- day_volume -SELECT TRUNC(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY TRUNC(paid_at) ORDER BY d; +SELECT TRUNC(paid_at) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '2026-10-02' GROUP BY TRUNC(paid_at) ORDER BY d; -- sample SELECT * FROM (SELECT status, COUNT(*) AS cnt FROM sales.orders SAMPLE BLOCK (1) WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status) WHERE ROWNUM <= 21; -- max_date diff --git a/backend/tests/fixtures/dialect/postgresql.sql b/backend/tests/fixtures/dialect/postgresql.sql index 995eb3f3..0a9b9ca4 100644 --- a/backend/tests/fixtures/dialect/postgresql.sql +++ b/backend/tests/fixtures/dialect/postgresql.sql @@ -22,7 +22,7 @@ SELECT MIN(CASE WHEN paid_at >= DATE '0001-01-02' AND paid_at < DATE '9999-12-31 -- date_window SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; -- day_volume -SELECT CAST(paid_at AS DATE) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '9999-12-31' GROUP BY CAST(paid_at AS DATE) ORDER BY d; +SELECT CAST(paid_at AS DATE) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= DATE '2026-08-27' AND paid_at < DATE '2026-10-02' GROUP BY CAST(paid_at AS DATE) ORDER BY d; -- sample SELECT status, COUNT(*) AS cnt FROM sales.orders TABLESAMPLE SYSTEM (1) WHERE status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; -- max_date diff --git a/backend/tests/fixtures/dialect/sqlite.sql b/backend/tests/fixtures/dialect/sqlite.sql index 44e0a904..1947d6d2 100644 --- a/backend/tests/fixtures/dialect/sqlite.sql +++ b/backend/tests/fixtures/dialect/sqlite.sql @@ -22,7 +22,7 @@ SELECT MIN(CASE WHEN paid_at >= '0001-01-02' AND paid_at < '9999-12-31' THEN pai -- date_window SELECT status, COUNT(*) AS cnt FROM sales.orders WHERE paid_at >= '2026-08-27' AND paid_at < '9999-12-31' AND status IS NOT NULL GROUP BY status ORDER BY cnt DESC, status LIMIT 21; -- day_volume -SELECT substr(paid_at, 1, 10) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= '2026-08-27' AND paid_at < '9999-12-31' GROUP BY substr(paid_at, 1, 10) ORDER BY d; +SELECT substr(paid_at, 1, 10) AS d, COUNT(*) AS n FROM sales.orders WHERE paid_at >= '2026-08-27' AND paid_at < '2026-10-02' GROUP BY substr(paid_at, 1, 10) ORDER BY d; -- sample -- （这种方言没有这类查询） -- max_date diff --git a/backend/tests/test_catalog_observe.py b/backend/tests/test_catalog_observe.py index 68d07455..76abfad0 100644 --- a/backend/tests/test_catalog_observe.py +++ b/backend/tests/test_catalog_observe.py @@ -228,6 +228,49 @@ async def test_large_table_basis_window_or_first_rows_and_no_cross_basis_compare await drop_source(b) +async def test_future_business_dates_stay_out_of_the_daily_window(client, tmp_path): + """业务日期最晚到明天、今天只写了几行：按日行数的查询只数到今天（上界，今天以后的不算），不记量跌。以前窗口只有下界， + 明天那一天也进了窗口，比较器再退一天就退到今天这个不完整的日子上，记出一条量跌（下一轮验收）。""" + path = tmp_path / "ahead.db" + _build(path, rows=0) + db = sqlite3.connect(path) + rows = [(TODAY - timedelta(days=d)).isoformat() for d in range(1, 31) for _ in range(30)] + rows += [TODAY.isoformat()] * 3 + [(TODAY + timedelta(days=1)).isoformat()] * 30 + db.executemany("INSERT INTO orders (status, paid_on, amount, secret) VALUES ('paid', ?, 1, 'x')", [(r,) for r in rows]) + db.commit() + db.close() + sid = await _make(client, path) + try: + job_id = await _observe(sid) + got = {(o.kind, o.column_name): o for o in await _obs(sid)} + days = got[("daily_volume", "paid_on")].data["days"] + assert (TODAY + timedelta(days=1)).isoformat() not in days and days[TODAY.isoformat()] == 3 + assert got[("daily_volume", "paid_on")].data["until"] == TODAY.isoformat() + async with SessionLocal() as session: + drops = list((await session.execute(select(SourceChange).where( + SourceChange.source_id == sid, SourceChange.kind == "volume_drop"))).scalars()) + assert drops == [], "今天还没写完、明天的日子不是整天：不算量跌" + assert all(o.job_id == job_id for o in got.values()) + finally: + await drop_source(sid) + + +def test_observation_today_is_the_server_local_date(monkeypatch): + """观察用的「今天」按服务端所在时区（和待办汇总 notify.digest_at 同一个口径）：业务日期是本地日期，UTC 20 点在东八区 + 已经是第二天。""" + import time + from datetime import datetime, timezone + + monkeypatch.setenv("TZ", "Asia/Shanghai") + time.tzset() + try: + assert observe.local_today(datetime(2026, 10, 17, 20, 0, tzinfo=timezone.utc)) == date(2026, 10, 18) + assert observe.local_today(datetime(2026, 10, 17, 15, 0, tzinfo=timezone.utc)) == date(2026, 10, 17) + finally: + monkeypatch.undo() + time.tzset() + + class _FakeCx: """只看拼出来的 SQL：大表、业务日期没有索引。""" diff --git a/backend/tests/test_catalog_profile_units.py b/backend/tests/test_catalog_profile_units.py index 22933754..7ce3c114 100644 --- a/backend/tests/test_catalog_profile_units.py +++ b/backend/tests/test_catalog_profile_units.py @@ -145,7 +145,7 @@ def test_profile_sql_passes_readonly_guard_on_every_dialect(kind): d.value_counts_sql(tbl, col, 21, scan_cap=100_000), d.min_max_sql(tbl, [(col, False)]), d.min_max_sql(tbl, [(col, True)]), - d.day_volume_sql(tbl, col, "2026-08-01"), + d.day_volume_sql(tbl, col, "2026-08-01", "2026-09-04"), d.window_value_counts_sql(tbl, col, col, "2026-08-01", 21), ] sampled = d.sample_value_counts_sql(tbl, col, 1, 21) diff --git a/backend/tests/test_profile_dialect_golden.py b/backend/tests/test_profile_dialect_golden.py index ebdd546e..ffd9b366 100644 --- a/backend/tests/test_profile_dialect_golden.py +++ b/backend/tests/test_profile_dialect_golden.py @@ -69,7 +69,7 @@ def _queries(kind: str) -> list[tuple[str, str | None]]: (d.column(META["orders"], "biz_day"), True)])), ("date_window", d.window_value_counts_sql(orders, status, d.column(META["orders"], "paid_at"), "2026-08-27", 21)), - ("day_volume", d.day_volume_sql(orders, d.column(META["orders"], "paid_at"), "2026-08-27")), + ("day_volume", d.day_volume_sql(orders, d.column(META["orders"], "paid_at"), "2026-08-27", "2026-10-01")), ("sample", d.sample_value_counts_sql(orders, status, 1, 21)), # 第 3 波：取值观察的最新业务日期（文字存的日期按字符串比） ("max_date", d.max_date_sql(orders, d.column(META["orders"], "paid_at"))), diff --git a/backend/tests/test_value_drift.py b/backend/tests/test_value_drift.py index 6fa80ef9..9d45e446 100644 --- a/backend/tests/test_value_drift.py +++ b/backend/tests/test_value_drift.py @@ -98,6 +98,33 @@ def test_volume_drop_against_the_previous_28_days(): assert vd.daily(partial, [], today=TODAY) == [] +def test_future_business_dates_do_not_turn_today_into_the_last_full_day(): + """业务日期有未来值（最晚到明天）：最近一个整天仍是昨天，今天这个还没写完的日子不算量跌；昨天真跌了照样报。以前取 + max(days) 再退一天，max 在明天时就落到了今天（只有几行）上，记出一条量跌。""" + start = TODAY - timedelta(days=30) # start + 30 是今天 + ahead = _days(start, [100] * 30 + [5, 40]) # 昨天 100、今天 5 行、明天 40 行 + assert not [f for f in vd.daily(ahead, [], today=TODAY) if f.kind == "volume_drop"] + [f] = [f for f in vd.daily(_days(start, [100] * 29 + [20, 5, 40]), [], today=TODAY) if f.kind == "volume_drop"] + assert f.after["day"] == (TODAY - timedelta(days=1)).isoformat() and f.after["rows"] == 20 + + +def test_days_after_today_are_not_gaps_now_or_in_history(): + """缺天只看到今天为止：今天以后两天没数据、第三天又有（预约、计划类的日期）不算缺天；以前那次观察窗口里它自己的「今天」 + 以后的空日子也不算以前有过缺天，挡不住这一次真缺的天。""" + start = TODAY - timedelta(days=30) + full = _days(start - timedelta(days=1), [100] * 31) + ahead = _days(start, [100] * 31 + [0, 0, 3]) # 今天之后空两天、第三天 3 行 + assert not [f for f in vd.daily(ahead, [full], today=TODAY) if f.kind == "date_gap"] + then = TODAY - timedelta(days=7) + old = vd.Obs("daily_volume", "window:paid_on:35", { + "until": then.isoformat(), + "days": {**{(then - timedelta(days=i)).isoformat(): 100 for i in range(30)}, + (then + timedelta(days=5)).isoformat(): 2}}, AT) # 那次的「今天」以后空了几天、又有一行 + holed = _days(start, [100] * 10 + [0, 0] + [100] * 19) + [gap] = [f for f in vd.daily(holed, [old], today=TODAY) if f.kind == "date_gap"] + assert gap.after["gaps"] == 2 + + def _rows(n: int, method: str = "count") -> vd.Obs: return vd.Obs("rows", method, {"rows": n, "method": method}, AT) diff --git a/scripts/dialect_it.py b/scripts/dialect_it.py index 86538338..f5621d42 100644 --- a/scripts/dialect_it.py +++ b/scripts/dialect_it.py @@ -50,6 +50,9 @@ SETTINGS = ProfileSettings(enabled=True, table_max_queries=40, table_max_s=60, q N_ORDERS, N_EVENTS_OLD, N_EVENTS_NEW = 40, 10, 1500 BASE = _dt.datetime(2026, 9, 1, 10, 0, 0) WINDOW_FROM = "2026-09-05" +#: 按日行数的上界（观察当天，含）：WINDOW_UNTIL 在全部夹具日期之后；WINDOW_CUT 落在中间，之后的日子不进窗口 +WINDOW_UNTIL = "2026-09-30" +WINDOW_CUT = "2026-09-07" failures: list[str] = [] @@ -80,6 +83,8 @@ EXPECT = { "pay_way": Counter(o["pay_way"] for o in ORDERS), "paid_at": (min(o["paid_at"] for o in VALID), max(o["paid_at"] for o in VALID)), "day_volume": Counter(o["paid_at"].date().isoformat() for o in IN_WINDOW), + "day_volume_cut": Counter(o["paid_at"].date().isoformat() for o in IN_WINDOW + if o["paid_at"].date().isoformat() <= WINDOW_CUT), "window_status": Counter(o["status"] for o in IN_WINDOW), } @@ -354,9 +359,13 @@ async def run_target(t: Target) -> None: # ---- 取值观察的新方法 tbl = d.table(orders, "orders") paid_col = d.column(orders, _key(orders, "paid_at")) - res = await data_engine.run_query(source, d.day_volume_sql(tbl, paid_col, WINDOW_FROM)) + res = await data_engine.run_query(source, d.day_volume_sql(tbl, paid_col, WINDOW_FROM, WINDOW_UNTIL)) got_days = Counter({_day(row[0]): int(row[1]) for row in res.rows}) check(got_days == EXPECT["day_volume"], f"按日行数和手算一致（无效日期不算进窗口）：{dict(got_days)}") + res = await data_engine.run_query(source, d.day_volume_sql(tbl, paid_col, WINDOW_FROM, WINDOW_CUT)) + got_cut = Counter({_day(row[0]): int(row[1]) for row in res.rows}) + check(got_cut == EXPECT["day_volume_cut"], + f"按日行数只数到观察当天（含当天的时分秒，之后的不算）：{dict(got_cut)}") status_col = d.column(orders, _key(orders, "status")) res = await data_engine.run_query(source, d.window_value_counts_sql(tbl, status_col, paid_col, WINDOW_FROM, 21)) got_window = Counter({str(row[0]).rstrip(" "): int(row[1]) for row in res.rows})