mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:21:58 +00:00
取值观察：业务日期有未来值时不再误报量跌、缺天；「今天」统一按服务端所在时区
验收复现：业务日期最晚到明天、今天只写了几行，记出一条 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 加了一条「只数到观察 当天」的真库核对。
This commit is contained in:
ThearchyHelios
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
@@ -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：大表、业务日期没有索引。"""
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"))),
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
+10
-1
@@ -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})
|
||||
|
||||
Reference in New Issue
Block a user