Fatih Kadir Akın

446prompt
0lượt bình chọn nhận được
0đóng góp
Tham gia about 13 hours ago
446 đóng góp trong năm qua
Tháng 11
Tháng 12
Tháng 1
Tháng 2
Tháng 3
Tháng 4
Tháng 5
Tháng 6
Tháng 7
Tháng 8
Tháng 9
Tháng 10
Ít
Nhiều
Công cụ Kiểm tra Khoảng trống và Điểm bất thường trong Chuỗi thời gian
Kỹ năng
Kiểm tra dữ liệu chuỗi thời gian dạng CSV (số liệu đo lường, chỉ số cảm biến, doanh số theo giờ) để phát hiện các vấn đề dữ liệu trước khi vẽ biểu đồ hoặc thiết lập cảnh báo: thiếu khoảng thời gian, dấu thời gian (timestamp) bị trùng lặp hoặc sai thứ tự, giá trị trống, đường đi ngang (flatlines) do trình xuất dữ liệu bị treo, các điểm đột biến (spikes) phát hiện qua trung vị/MAD bền vững và hiện tượng đặt lại bộ đếm (counter resets). Sử dụng khi bảng điều khiển (dashboard) có dấu hiệu bất thường, trước khi huấn luyện mô hình dự báo hoặc khi cần trả lời câu hỏi "dữ liệu này đã đầy đủ chưa?". Bao gồm công cụ kiểm tra bằng thư viện chuẩn Python có hỗ trợ cờ --self-test.
---
name: time-series-gap-anomaly-checker
description: Kiểm tra dữ liệu chuỗi thời gian dạng CSV (số liệu đo lường, chỉ số cảm biến, doanh số theo giờ) để phát hiện các vấn đề dữ liệu trước khi vẽ biểu đồ hoặc thiết lập cảnh báo: thiếu khoảng thời gian, dấu thời gian (timestamp) bị trùng lặp hoặc sai thứ tự, giá trị trống, đường đi ngang (flatlines) do trình xuất dữ liệu bị treo, các điểm đột biến (spikes) phát hiện qua trung vị/MAD bền vững và hiện tượng đặt lại bộ đếm (counter resets). Sử dụng khi bảng điều khiển (dashboard) có dấu hiệu bất thường, trước khi huấn luyện mô hình dự báo hoặc khi cần trả lời câu hỏi "dữ liệu này đã đầy đủ chưa?". Bao gồm công cụ kiểm tra bằng thư viện chuẩn Python có hỗ trợ cờ --self-test.
---

# Công cụ Kiểm tra Khoảng trống và Điểm bất thường trong Chuỗi thời gian

Bạn giúp các chuyên viên phân tích, kỹ sư SRE và kỹ sư dữ liệu quyết định xem một chuỗi thời gian có đáng tin cậy hay không. Bạn tìm ra những vị trí dữ liệu bị thiếu, lặp lại, bị treo cứng hoặc đáng ngờ, giải thích nguyên nhân khả dĩ của từng vấn đề, đồng thời tách biệt các sự cố thu thập dữ liệu (trên đường ống pipeline) khỏi các sự kiện thực tế (trên hệ thống đang được đo lường). Bạn không bao giờ âm thầm tự lấp đầy các khoảng trống hoặc xóa bỏ các điểm ngoại lai (outliers); bạn đề xuất phương án xử lý và nêu rõ phương án đó làm thay đổi điều gì.

## Các tệp trong kỹ năng này

- `scripts/ts_check.py` - đọc tệp CSV, suy luận khoảng thời gian lấy mẫu (sampling interval) và báo cáo các khoảng trống, dữ liệu trùng lặp, các dòng sai thứ tự, giá trị bị thiếu, đường đi ngang (flatlines), điểm đột biến (spikes) và hiện tượng đặt lại bộ đếm dưới dạng markdown hoặc JSON (chỉ sử dụng thư viện chuẩn Python 3)
- `references/anomaly-methods.md` - cách thức hoạt động của các phép kiểm tra, cách tinh chỉnh ngưỡng (thresholds) và những trường hợp mỗi phương pháp không phát huy tác dụng
- `references/gap-handling-playbook.md` - cách xử lý cho từng loại vấn đề: điền giá trị (fill), nội suy (interpolate), loại bỏ (drop) hoặc ghi chú (annotate)
- `templates/data-quality-report.md` - mẫu báo cáo phục vụ việc sửa lỗi dashboard, đánh giá sự cố hoặc bàn giao bộ dữ liệu
- `examples/api-metrics.csv` - dữ liệu số liệu API mỗi 5 phút trong suốt ba giờ, bao gồm một đợt gián đoạn thu thập dữ liệu, một dòng trùng lặp, một giá trị bị treo, một điểm đột biến độ trễ và một lần đặt lại bộ đếm

## Quy trình thực hiện

### 1. Tìm hiểu chuỗi dữ liệu
Hỏi xem mỗi cột đo lường đại lượng gì, đơn vị tính, khoảng thời gian lấy mẫu kỳ vọng là bao nhiêu, và liệu có cột nào là bộ đếm đơn điệu (monotonic counter) hay không (ví dụ: requests_total, bytes_sent). Hỏi xem các dấu thời gian không kèm múi giờ (naive timestamps) đang sử dụng múi giờ nào; kịch bản mặc định coi chúng là giờ UTC.

### 2. Chạy công cụ kiểm tra
```
python3 scripts/ts_check.py examples/api-metrics.csv --counter requests_total
python3 scripts/ts_check.py data.csv --time-col ts --cols temp_c,humidity --interval 10m
python3 scripts/ts_check.py data.csv --z 5 --window 12 --flatline-points 12 --json > report.json
python3 scripts/ts_check.py --self-test
```
Dấu thời gian có thể ở định dạng ISO 8601 hoặc thời gian Unix epoch tính bằng giây hoặc mili-giây. Trả về mã thoát (exit code) là 1 khi phát hiện bất kỳ vấn đề nào, và 2 khi đầu vào không hợp lệ, nhờ đó có thể dùng làm cổng kiểm soát (gate) trong đường ống dữ liệu.

### 3. Đọc kết quả
- **Độ đầy đủ (Completeness)** = số dòng hiện có / số dòng kỳ vọng trên toàn dải thời gian. Mức dưới 95% thường đồng nghĩa với việc các con số tổng hợp và trung bình đã bị sai lệch.
- **gap**: khoảng cách giữa các điểm vượt quá 1.5 lần khoảng lấy mẫu. Các khoảng trống dài xuất hiện vào những khung giờ cố định thường là do quá trình triển khai (deploy) hoặc tác vụ định kỳ (cron job); một khoảng trống dài đơn lẻ thường do bộ thu thập dữ liệu (collector) bị ngừng hoạt động.
- **duplicate / out_of_order**: thu thập dữ liệu bị đúp, cơ chế thử lại (retries) hoặc do gộp nhiều tệp với nhau.
- **flatline**: cùng một giá trị lặp đi lặp lại; đây là điều bình thường đối với tín hiệu bật/tắt (on/off), nhưng lại rất đáng ngờ đối với các phép đo vốn có độ nhiễu tự nhiên.
- **spike**: điểm z-score bền vững (robust z-score) so với trung vị trượt (rolling median). Hãy kiểm tra xem các cột lân cận có tăng đột biến cùng lúc hay không (nếu có thì là sự kiện thực tế, nếu không thì là lỗi nhiễu dữ liệu).
- **reset**: bộ đếm bị sụt giảm giá trị, thường do tiến trình khởi động lại; các phép tính tốc độ thay đổi (rate) bắt buộc phải xử lý trường hợp này.

### 4. Đề xuất phương án xử lý
Sử dụng `references/gap-handling-playbook.md`. Tinh chỉnh các ngưỡng bằng `references/anomaly-methods.md` nếu dữ liệu có tính mùa vụ, biến động theo từng đợt (bursty) hoặc rất thưa thớt.

### 5. Báo cáo
Điền vào mẫu `templates/data-quality-report.md` với các con số tóm tắt, những vấn đề hàng đầu kèm theo thời gian theo múi giờ của người dùng, nguyên nhân nghi ngờ và phương án xử lý đã thống nhất.

## Quy tắc

- Chỉ báo cáo các vấn đề; không tự ý sửa đổi dữ liệu nguồn. Ghi dữ liệu đã làm sạch ra một tệp mới và ghi lại tài liệu cho mọi thay đổi đã thực hiện.
- Không kết luận một điểm đột biến (spike) là sự cố hệ thống nếu chưa có bằng chứng củng cố (từ các số liệu khác, nhật ký hệ thống hoặc xác nhận của con người).
- Nêu rõ khoảng lấy mẫu và các ngưỡng bạn đã sử dụng để kết quả có thể tái lập lại được.
- Chuyển đổi giờ UTC sang múi giờ của người dùng khi trình bày kết quả.
FILE:references/anomaly-methods.md
# Phương pháp Phát hiện Bất thường và Tinh chỉnh Tham số

## Suy luận khoảng thời gian lấy mẫu
Khoảng thời gian lấy mẫu kỳ vọng là trung vị của khoảng cách giữa các dấu thời gian duy nhất đã được sắp xếp tăng dần. Cách tính này vẫn chính xác ngay cả khi có một vài khoảng trống và độ trễ dao động nhỏ (jitter). Hãy truyền tham số `--interval` khi chuỗi dữ liệu phần lớn là khoảng trống (dữ liệu sự kiện thưa thớt) hoặc khi bạn đã biết chắc chắn chu kỳ thu thập dữ liệu.

Khoảng trống (gap) là bất kỳ khoảng cách nào lớn hơn 1.5 lần khoảng lấy mẫu; vì vậy với dữ liệu 5 phút một lần, khoảng cách 7 phút không bị coi là khoảng trống nhưng khoảng cách 10 phút thì được tính là một khoảng trống (thiếu 1 điểm dữ liệu).

## Phát hiện điểm đột biến bền vững (trung vị trượt + MAD)
Đối với mỗi điểm, lấy tối đa `--window` điểm lân cận ở mỗi bên (không tính chính điểm đang xét), tính trung vị (median) và độ lệch tuyệt đối trung vị (MAD - median absolute deviation) của chúng. Điểm z-score bền vững được tính theo công thức:

    z = |value - median| / (1.4826 * MAD)

Hệ số 1.4826 giúp quy đổi MAD tương đương với độ lệch chuẩn đối với dữ liệu phân phối chuẩn. Ngưỡng mặc định bằng 4 tương ứng với xác suất khoảng 1 trên 15.000 đối với nhiễu phân phối chuẩn. Khi có hơn một nửa số điểm lân cận mang giá trị giống hệt nhau (MAD = 0), độ lệch tuyệt đối trung bình (mean absolute deviation) sẽ được sử dụng thay thế, và nếu toàn bộ vùng lân cận hoàn toàn phẳng thì lấy bằng 1% mức giá trị đó.

Tại sao không dùng giá trị trung bình (mean) và độ lệch chuẩn (standard deviation)? Chỉ cần một điểm đột biến lớn cũng đủ làm phình đại độ lệch chuẩn và khiến chính điểm đột biến đó bị che khuất. Ngược lại, trung vị và MAD hoàn toàn miễn nhiễm với hiện tượng này.

## Tinh chỉnh tham số
| Loại dữ liệu | Gợi ý cấu hình |
|---|---|
| Số liệu nhiều nhiễu (độ trễ, CPU) | `--z 5`, `--window 12` |
| Có tính mùa vụ theo ngày hoặc tuần | Kiểm tra từng chu kỳ mùa vụ một, hoặc so sánh với cùng khung giờ tuần trước trước khi tin vào các điểm đột biến |
| Dữ liệu kinh doanh tăng vọt theo đợt (đơn hàng) | Tăng `--z` cao hơn (6-8); các điểm vọt lên thường là sự kiện thật |
| Giá trị số nguyên hoặc độ phân giải thấp | Thường gặp MAD = 0; hãy tăng `--z` hoặc tổng hợp (aggregate) dữ liệu trước |
| Tín hiệu bật/tắt hoặc trạng thái | Bỏ qua cảnh báo flatline; chỉ kiểm tra các khoảng trống (gaps) |
| Bộ đếm (Counters) | Luôn truyền cờ `--counter`; kiểm tra tốc độ biến thiên (rates) sau khi đã xử lý các điểm reset |

## Các giới hạn đã biết
- Sự trượt giá trị từ từ (gradual drifts) và sự dịch chuyển nền giá trị (level shifts) không phải là điểm đột biến ngắn hạn; hãy vẽ biểu đồ chuỗi dữ liệu hoặc so sánh trung vị giữa các tuần.
- Một vài điểm đầu tiên và cuối cùng có ít điểm lân cận hơn, do đó điểm số của chúng kém tin cậy hơn.
- Các luồng sự kiện không đều đặn (lượt đăng nhập, lượt mua hàng) không có khoảng lấy mẫu cố định; hãy gom nhóm (aggregate) chúng vào các khung thời gian (buckets) trước.
FILE:references/gap-handling-playbook.md
# Cẩm nang Xử lý Khoảng trống và Điểm bất thường

| Vấn đề | Nguyên nhân khả dĩ | Phương án xử lý mặc định | Điều cần tránh |
|---|---|---|---|
| Khoảng trống ngắn (1-2 điểm) | quá thời gian thu thập (scrape timeout), chập chờn mạng thoáng qua | Nội suy tuyến tính cho các tín hiệu mượt; giữ nguyên giá trị null đối với số đếm | Điền số 0 vào các cột số đếm |
| Khoảng trống dài | trình thu thập ngừng chạy, đang triển khai hệ thống, máy chủ tắt | Giữ nguyên null và ghi chú lên biểu đồ; loại trừ khỏi phép tính SLO hoặc tính là trạng thái chưa xác định (unknown) | Nội suy bắc cầu qua nhiều giờ liền |
| Dấu thời gian trùng lặp | thu thập đúp, chạy lại lô dữ liệu (retried batch), gộp nhiều tệp | Giữ lại một dòng duy nhất (dòng đầu tiên hoặc dòng có nhiều trường dữ liệu nhất); kiểm tra xem các giá trị có khác nhau không | Cộng gộp các dòng trùng lặp |
| Dòng dữ liệu sai thứ tự thời gian | dữ liệu đến muộn, nối ghép các tệp | Sắp xếp lại theo thời gian trước khi thực hiện bất kỳ phép tính trượt (windowed calculation) nào | Mặc định thứ tự dòng trong tệp là thứ tự thời gian |
| Giá trị bị thiếu (Missing values) | cảm biến mất tín hiệu tạm thời, lỗi phân tích cú pháp, chuỗi "N/A" | Sửa lỗi bộ phân tích cú pháp (parser) trước; sau đó xử lý như một khoảng trống | Coi ô trống là giá trị 0 |
| Đường đi ngang (Flatline) | trình xuất dữ liệu bị treo, giá trị lấy từ bộ nhớ đệm, cảm biến bị đơ | Xác nhận chéo với nguồn dữ liệu thứ hai; đánh dấu là đáng ngờ | Tin tưởng dữ liệu này để kích hoạt cảnh báo |
| Điểm đột biến (Spike đơn lẻ) | lỗi nhiễu thoáng qua, sai đơn vị đo, đợt tăng tải thực tế | Kiểm tra các cột liên quan và nhật ký hệ thống; chỉ giới hạn trần (cap) hoặc loại bỏ khi có lý do được ghi chép rõ ràng | Xóa bỏ mà không để lại ghi chú |
| Đặt lại bộ đếm (Counter reset) | tiến trình khởi động lại, tràn bộ đếm (overflow) | Tính tốc độ (rate) theo từng phân đoạn (tốc độ sau khi reset = giá trị mới / khoảng thời gian) | Để xuất hiện tốc độ âm |

## Trước khi xây dựng mô hình hoặc dự báo
1. Sắp xếp, khử trùng lặp và chuẩn hóa về đúng khoảng thời gian lấy mẫu kỳ vọng (chèn giá trị null vào các vị trí bị khuyết).
2. Quyết định chính sách xử lý khoảng trống cho từng cột và ghi nhận lại.
3. Duy trì thêm một cột đánh dấu mặt nạ (`is_imputed`) để mô hình và người đọc biết những vị trí nào đã được điền dữ liệu nhân tạo.
4. Chạy lại công cụ kiểm tra trên tệp đã làm sạch và đính kèm cả hai bản báo cáo.

## Khi nào cần báo cáo nâng cấp độ ưu tiên (Escalate)
- Độ đầy đủ (Completeness) dưới 90% đối với một chỉ số được dùng để tính SLO hoặc lập hóa đơn thanh toán.
- Các khoảng trống lặp đi lặp lại vào cùng một thời điểm mỗi ngày (có một tác vụ ngầm nào đó đang làm sập bộ thu thập dữ liệu).
- Các điểm đột biến xuất hiện đồng loạt trên tất cả các máy chủ cùng một lúc (đây là sự cố hệ thống thực sự, không phải lỗi dữ liệu).
FILE:templates/data-quality-report.md
# Chất lượng Dữ liệu Chuỗi thời gian - <tên bộ dữ liệu hoặc dashboard>

**Nguồn:** <tệp / truy vấn>  **Phạm vi:** <bắt đầu> đến <kết thúc> (<múi giờ>)  **Khoảng lấy mẫu:** <kỳ vọng>
**Số dòng:** <n>  **Độ đầy đủ:** <phần trăm>  **Ngưỡng áp dụng:** z=<z>, window=<w>, flatline=<n>

## Tóm tắt
<hai câu: dữ liệu này có đáng tin cậy cho mục đích sử dụng dự kiến hay không, và điều gì cần phải thay đổi trước tiên>

## Các vấn đề phát hiện
| Loại vấn đề | Thời điểm (<múi giờ>) | Cột | Chi tiết | Nguyên nhân nghi ngờ | Phương án xử lý |
|---|---|---|---|---|---|
| | | | | | |

## Các quyết định xử lý
- Chính sách xử lý khoảng trống: <nội suy các khoảng trống ngắn / giữ nguyên null / loại trừ>
- Chính sách xử lý điểm ngoại lai: <giữ nguyên / giới hạn trần / loại bỏ kèm ghi chú>
- Xử lý bộ đếm: <tính tốc độ theo phân đoạn tại các điểm reset>

## Công việc tiếp theo
- [ ] <sửa bộ thu thập dữ liệu / bộ phân tích cú pháp / cảnh báo> - người phụ trách <tên> - hạn hoàn thành <ngày>

Được tạo bằng scripts/ts_check.py. Dữ liệu nguồn không bị sửa đổi.
FILE:examples/api-metrics.csv
timestamp,p95_latency_ms,error_rate,requests_total
2026-10-18T22:00:00Z,182,0.4,1200
2026-10-18T22:05:00Z,176,0.3,2435
2026-10-18T22:10:00Z,190,0.5,3705
2026-10-18T22:15:00Z,185,0.4,5010
2026-10-18T22:20:00Z,179,0.3,6210
2026-10-18T22:25:00Z,188,0.4,7445
2026-10-18T22:30:00Z,181,0.5,8715
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:35:00Z,193,0.4,10020
2026-10-18T22:40:00Z,177,0.3,11220
2026-10-18T22:45:00Z,184,0.4,12455
2026-10-18T22:50:00Z,186,0.4,13725
2026-10-18T22:55:00Z,180,0.4,15030
2026-10-18T23:00:00Z,191,0.4,16230
2026-10-18T23:05:00Z,,0.4,17465
2026-10-18T23:10:00Z,178,0.4,18735
2026-10-18T23:15:00Z,187,0.4,20040
2026-10-18T23:20:00Z,182,0.4,21240
2026-10-18T23:25:00Z,189,0.4,22475
2026-10-18T23:30:00Z,612,2.1,23745
2026-10-18T23:35:00Z,185,0.5,25050
2026-10-18T23:40:00Z,180,0.3,26250
2026-10-18T23:45:00Z,184,0.4,27485
2026-10-19T00:10:00Z,188,0.5,28755
2026-10-19T00:15:00Z,183,0.4,30060
2026-10-19T00:20:00Z,177,0.3,31260
2026-10-19T00:25:00Z,185,0.4,32495
2026-10-19T00:30:00Z,182,0.5,1180
2026-10-19T00:35:00Z,189,0.4,2485
2026-10-19T00:40:00Z,184,0.3,3685
2026-10-19T00:45:00Z,180,0.4,4920
2026-10-19T00:50:00Z,186,0.5,6190
2026-10-19T00:55:00Z,183,0.4,7495
FILE:scripts/ts_check.py
#!/usr/bin/env python3
"""Time series gap and anomaly checker for CSV metrics (Python 3 standard library only).

Reads a CSV with a timestamp column and one or more numeric columns, infers the sampling
interval (median spacing) unless --interval is given, and reports:
  gap          missing samples (spacing > 1.5 x interval), with how many points are missing
  duplicate    repeated timestamps
  out_of_order timestamps that go backwards
  missing      empty, NaN, or non-numeric values
  flatline     the same value repeated for >= --flatline-points samples (stuck sensor/exporter)
  spike        robust outliers: |value - rolling median| / (1.4826 * MAD) >= --z (default 4);
               if MAD is 0 the mean absolute deviation is used instead
  reset        a counter column that decreases (--counter col1,col2)

Timestamps: ISO 8601 (with Z or offset; naive = UTC) or Unix epoch seconds/milliseconds.

Usage:
  python3 ts_check.py metrics.csv
  python3 ts_check.py metrics.csv --time-col ts --cols cpu,latency_ms --interval 5m --z 5
  python3 ts_check.py metrics.csv --counter requests_total --json
  python3 ts_check.py --self-test
Exit code: 0 clean, 1 issues found, 2 bad input.
"""
import argparse
import csv
import datetime as dt
import io
import json
import math
import re
import statistics
import sys


class InputError(ValueError):
    pass


def parse_ts(raw):
    s = (raw or "").strip()
    if not s:
        raise InputError("empty timestamp")
    if re.fullmatch(r"-?\d+(\.\d+)?", s):
        v = float(s)
        return v / 1000.0 if v > 1e11 else v
    try:
        t = dt.datetime.fromisoformat(s.replace("Z", "+00:00").replace(" ", "T", 1))
    except ValueError:
        raise InputError(f"unparseable timestamp {s!r}")
    if t.tzinfo is None:
        t = t.replace(tzinfo=dt.timezone.utc)
    return t.timestamp()


def parse_interval(s):
    m = re.fullmatch(r"(\d+(?:\.\d+)?)\s*(ms|s|m|h|d)?", s.strip())
    if not m:
        raise InputError(f"bad interval {s!r}; use e.g. 30s, 5m, 1h")
    mult = {"ms": 0.001, "s": 1, None: 1, "m": 60, "h": 3600, "d": 86400}[m.group(2)]
    return float(m.group(1)) * mult


def fmt_ts(ts):
    return dt.datetime.fromtimestamp(ts, dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def fmt_dur(sec):
    for unit, size in (("d", 86400), ("h", 3600), ("m", 60)):
        if sec >= size:
            v = sec / size
            return f"{v:.0f}{unit}" if v == int(v) else f"{v:.1f}{unit}"
    return f"{sec:g}s"


def to_float(raw):
    s = (raw or "").strip()
    if s == "" or s.lower() in ("nan", "null", "none", "na", "n/a", "-"):
        return None
    try:
        v = float(s)
    except ValueError:
        return None
    return None if math.isnan(v) or math.isinf(v) else v


def load(text, time_col=None, cols=None):
    reader = csv.DictReader(io.StringIO(text))
    if not reader.fieldnames:
        raise InputError("CSV has no header row")
    fields = [f.strip() for f in reader.fieldnames]
    reader.fieldnames = fields
    if time_col is None:
        cands = [f for f in fields if f.lower() in ("timestamp", "time", "ts", "date", "datetime")]
        time_col = cands[0] if cands else fields[0]
    if time_col not in fields:
        raise InputError(f"time column {time_col!r} not in header {fields}")
    value_cols = cols or [f for f in fields if f != time_col]
    for c in value_cols:
        if c not in fields:
            raise InputError(f"column {c!r} not in header {fields}")
    rows = []
    for i, r in enumerate(reader, start=2):
        rows.append({"line": i, "ts": parse_ts(r.get(time_col)),
                     "raw": {c: r.get(c) for c in value_cols}})
    if len(rows) < 3:
        raise InputError("need at least 3 data rows")
    return time_col, value_cols, rows


def rolling_robust(values, idx, window):
    lo, hi = max(0, idx - window), min(len(values), idx + window + 1)
    neigh = [values[j] for j in range(lo, hi) if j != idx and values[j] is not None]
    if len(neigh) < 4:
        return None, None
    med = statistics.median(neigh)
    mad = statistics.median(abs(v - med) for v in neigh)
    scale = 1.4826 * mad
    if scale == 0:  # more than half the neighbours equal the median: fall back to mean absolute deviation
        scale = 1.2533 * statistics.fmean(abs(v - med) for v in neigh)
    if scale == 0:  # perfectly flat neighbourhood: 1% of the level
        scale = max(abs(med) * 0.01, 1e-9)
    return med, scale


def check(text, time_col=None, cols=None, interval=None, z=4.0, window=6, flatline_points=6, counters=()):
    time_col, value_cols, rows = load(text, time_col, cols)
    issues = []

    def add(kind, line, ts, col, detail):
        issues.append({"kind": kind, "line": line, "time": fmt_ts(ts), "column": col, "detail": detail})

    # order and duplicates (in file order)
    seen = {}
    for prev, cur in zip(rows, rows[1:]):
        if cur["ts"] < prev["ts"]:
            add("out_of_order", cur["line"], cur["ts"], time_col, f"goes back {fmt_dur(prev['ts'] - cur['ts'])} from line {prev['line']}")
    for r in rows:
        if r["ts"] in seen:
            add("duplicate", r["line"], r["ts"], time_col, f"same timestamp as line {seen[r['ts']]}")
        else:
            seen[r["ts"]] = r["line"]
    srt = sorted({r["ts"]: r for r in reversed(rows)}.values(), key=lambda r: r["ts"])  # first occurrence wins
    diffs = [b["ts"] - a["ts"] for a, b in zip(srt, srt[1:]) if b["ts"] > a["ts"]]
    if interval is None:
        if not diffs:
            raise InputError("cannot infer interval: all timestamps identical")
        interval = statistics.median(diffs)
    expected = int(round((srt[-1]["ts"] - srt[0]["ts"]) / interval)) + 1
    missing_points = 0
    for a, b in zip(srt, srt[1:]):
        d = b["ts"] - a["ts"]
        if d > 1.5 * interval:
            n = int(round(d / interval)) - 1
            missing_points += n
            add("gap", b["line"], a["ts"], time_col, f"{fmt_dur(d)} without data after {fmt_ts(a['ts'])} (~{n} missing)")

    stats = {}
    for c in value_cols:
        vals = [to_float(r["raw"][c]) for r in srt]
        for r, v in zip(srt, vals):
            if v is None:
                add("missing", r["line"], r["ts"], c, f"value {r['raw'][c]!r} is empty or not numeric")
        # flatline runs
        run_start = 0
        for i in range(1, len(vals) + 1):
            if i < len(vals) and vals[i] is not None and vals[i] == vals[run_start]:
                continue
            length = i - run_start
            if vals[run_start] is not None and length >= flatline_points:
                r0, r1 = srt[run_start], srt[i - 1]
                add("flatline", r0["line"], r0["ts"], c, f"value {vals[run_start]:g} repeated {length}x until {fmt_ts(r1['ts'])}")
            run_start = i
        if c in counters:
            for i in range(1, len(vals)):
                if vals[i] is not None and vals[i - 1] is not None and vals[i] < vals[i - 1]:
                    add("reset", srt[i]["line"], srt[i]["ts"], c, f"counter dropped {vals[i - 1]:g} -> {vals[i]:g} (restart or wrap)")
        else:
            for i, v in enumerate(vals):
                if v is None:
                    continue
                med, scale = rolling_robust(vals, i, window)
                if med is None:
                    continue
                score = abs(v - med) / scale
                if score >= z:
                    add("spike", srt[i]["line"], srt[i]["ts"], c, f"{v:g} vs local median {med:g} (robust z {score:.1f})")
        good = [v for v in vals if v is not None]
        stats[c] = {"count": len(good), "min": min(good) if good else None, "max": max(good) if good else None,
                    "median": statistics.median(good) if good else None}

    order = {"out_of_order": 0, "duplicate": 1, "gap": 2, "reset": 3, "missing": 4, "flatline": 5, "spike": 6}
    issues.sort(key=lambda x: (order[x["kind"]], x["time"], x["column"]))
    counts = {}
    for it in issues:
        counts[it["kind"]] = counts.get(it["kind"], 0) + 1
    return {
        "rows": len(rows), "time_column": time_col, "columns": value_cols,
        "start": fmt_ts(srt[0]["ts"]), "end": fmt_ts(srt[-1]["ts"]),
        "interval": fmt_dur(interval), "interval_seconds": interval,
        "expected_points": expected, "missing_points": missing_points,
        "completeness": round(100.0 * len(srt) / expected, 1) if expected else 100.0,
        "stats": stats, "counts": counts, "issues": issues,
    }


def to_markdown(r):
    lines = ["# Time series check", "",
             f"Rows: {r['rows']}  Range: {r['start']} to {r['end']}  Interval: {r['interval']}",
             f"Completeness: {r['completeness']}% ({r['missing_points']} of {r['expected_points']} expected points missing)", "",
             "| Column | Count | Min | Median | Max |", "|---|---|---|---|---|"]
    for c, s in r["stats"].items():
        f = lambda v: "-" if v is None else f"{v:g}"
        lines.append(f"| {c} | {s['count']} | {f(s['min'])} | {f(s['median'])} | {f(s['max'])} |")
    lines += ["", "## Issues"]
    if not r["issues"]:
        lines.append("No issues found.")
    else:
        lines.append("Totals: " + ", ".join(f"{k} {v}" for k, v in r["counts"].items()))
        lines += ["", "| Kind | Time (UTC) | Column | Line | Detail |", "|---|---|---|---|---|"]
        lines += [f"| {i['kind']} | {i['time']} | {i['column']} | {i['line']} | {i['detail']} |" for i in r["issues"]]
    return "\n".join(lines)


def self_test():
    cases = []

    def ok(name, cond):
        cases.append((name, bool(cond)))

    base = 1760860800  # 2025-10-19T08:00:00Z
    def csv_of(points, header="timestamp,value"):
        return header + "\n" + "\n".join(f"{t},{v}" for t, v in points) + "\n"
    wave = [10 + (i % 3) for i in range(30)]
    clean = csv_of([(base + 60 * i, v) for i, v in enumerate(wave)])
    r = check(clean)
    ok("clean series has no issues", r["issues"] == [])
    ok("interval inferred as 1m", r["interval"] == "1m")
    ok("completeness 100%", r["completeness"] == 100.0)
    gap_pts = [(base + 60 * i, v) for i, v in enumerate(wave) if not 10 <= i < 15]
    r = check(csv_of(gap_pts))
    ok("gap detected", r["counts"].get("gap") == 1)
    ok("gap counts 5 missing points", r["missing_points"] == 5)
    spiky = list(wave); spiky[20] = 95
    r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(spiky)]))
    ok("spike detected", [i["line"] for i in r["issues"] if i["kind"] == "spike"] == [22])
    flat = list(wave); flat[5:13] = [7] * 8
    r = check(csv_of([(base + 60 * i, v) for i, v in enumerate(flat)]), z=100)
    ok("flatline detected", r["counts"].get("flatline") == 1)
    miss = [(base + 60 * i, ("" if i == 4 else "NaN" if i == 9 else v)) for i, v in enumerate(wave)]
    ok("missing values detected", check(csv_of(miss)).get("counts").get("missing") == 2)
    dup = [(base + 60 * i, v) for i, v in enumerate(wave)]
    dup.insert(6, dup[5])
    ok("duplicate detected", check(csv_of(dup))["counts"].get("duplicate") == 1)
    ooo = [(base + 60 * i, v) for i, v in enumerate(wave)]
    ooo[7], ooo[8] = ooo[8], ooo[7]
    ok("out of order detected", check(csv_of(ooo))["counts"].get("out_of_order") == 1)
    counter = [(base + 60 * i, 100 * i if i < 20 else 100 * (i - 20)) for i in range(30)]
    r = check(csv_of(counter, "ts,requests_total"), counters=("requests_total",))
    ok("counter reset detected", r["counts"].get("reset") == 1 and "spike" not in r["counts"])
    iso_rows = "time,cpu\n" + "\n".join(f"2026-10-19T08:{i:02d}:00Z,{40 + (i % 2)}" for i in range(0, 60, 5)) + "\n"
    r = check(iso_rows)
    ok("ISO timestamps and 5m interval", r["interval"] == "5m" and r["issues"] == [])
    ms_rows = csv_of([((base + 30 * i) * 1000, v) for i, v in enumerate(wave)])
    ok("epoch milliseconds accepted", check(ms_rows)["interval"] == "30s")
    r = check(clean, interval=30.0)
    ok("explicit 30s interval flags every 1m step as a gap", r["counts"].get("gap") == 29 and r["completeness"] < 60)
    jitter = csv_of([(base + 60 * i + (3 if i % 2 else -2), v) for i, v in enumerate(wave)])
    ok("small timestamp jitter is not a gap", "gap" not in check(jitter)["counts"])
    multi = "timestamp,a,b\n" + "\n".join(f"{base + 60 * i},{wave[i]},{wave[i] * 2 if i != 12 else 500}" for i in range(30)) + "\n"
    r = check(multi, cols=["a", "b"])
    ok("multi-column spike only in b", {i["column"] for i in r["issues"]} == {"b"})
    for bad, label in (("timestamp,value\n1,2\n", "too few rows rejected"),
                       ("timestamp,value\nyesterday,1\ntoday,2\nnow,3\n", "bad timestamp rejected")):
        try:
            check(bad)
            ok(label, False)
        except InputError:
            ok(label, True)
    ok("markdown renders", "## Issues" in to_markdown(check(clean)))
    passed = sum(c for _, c in cases)
    for name, c in cases:
        print(("PASS " if c else "FAIL ") + name)
    print(f"{passed}/{len(cases)} passed")
    return 0 if passed == len(cases) else 1


def main(argv=None):
    ap = argparse.ArgumentParser(description="Find gaps, duplicates, flatlines, spikes, and counter resets in CSV metrics.")
    ap.add_argument("csv", nargs="?", help="CSV file, or '-' for stdin")
    ap.add_argument("--time-col")
    ap.add_argument("--cols", help="comma-separated value columns (default: all except the time column)")
    ap.add_argument("--interval", help="expected spacing, e.g. 30s, 5m, 1h (default: median spacing)")
    ap.add_argument("--z", type=float, default=4.0, help="robust z-score threshold for spikes (default 4)")
    ap.add_argument("--window", type=int, default=6, help="neighbours on each side for the rolling median (default 6)")
    ap.add_argument("--flatline-points", type=int, default=6)
    ap.add_argument("--counter", default="", help="comma-separated monotonic counter columns (checked for resets, not spikes)")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--self-test", action="store_true")
    a = ap.parse_args(argv)
    if a.self_test:
        return self_test()
    if not a.csv:
        ap.error("csv path required (or --self-test)")
    try:
        text = sys.stdin.read() if a.csv == "-" else open(a.csv, encoding="utf-8-sig").read()
        r = check(text, a.time_col, [c.strip() for c in a.cols.split(",")] if a.cols else None,
                  parse_interval(a.interval) if a.interval else None, a.z, a.window, a.flatline_points,
                  tuple(c.strip() for c in a.counter.split(",") if c.strip()))
    except (OSError, InputError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    print(json.dumps(r, indent=2) if a.json else to_markdown(r))
    return 1 if r["issues"] else 0


if __name__ == "__main__":
    sys.exit(main())