fix(stats): reconcile stale gps mileage candidates

2026-07-20 03:51:18 +08:00
parent 5ba636419b
commit 758da369d8
6 changed files with 172 additions and 5 deletions
--- a/docs/ops/2026-07-20-daily-mileage-track-reconciliation.md
+++ b/docs/ops/2026-07-20-daily-mileage-track-reconciliation.md
@@ -56,6 +56,8 @@ stat-writer 已支持在累计里程缺失时，用有速度证据的相邻 GPS

 后续历史补录继续使用同一条安全规则：只有存在可用移动线段时才生成 `gps_coordinate_accumulation` 候选，不按累计总里程、位置点数量或页面展示值推断日里程。

+2026-07-20 03:44 对当天三协议再次执行全量 dry-run，共重建 448 个来源聚合，识别 5 个可由 JT808 有效移动轨迹恢复的候选；03:48 正式回补写入 449 个来源聚合并归一化 272 个 JT808 平台来源。复核同时发现阈值规则上线前遗留的沪A56988F `0.004 km` GPS 候选，因此补充了“低于 100 米遗留候选清理 + 同事务重新投影”能力；后续回补会自动修复同类旧数据，而不是只阻止新漂移值写入。
+
 ## 限制与后续监控

 - GPS 估算只在仪表里程缺失或停滞时兜底，不替代 GB32960、宇通仪表里程或 JT808 设备里程。
--- a/docs/ops/vehicle-ingest-runbook.md
+++ b/docs/ops/vehicle-ingest-runbook.md
@@ -511,7 +511,7 @@ BACKFILL_DATE_TO=2026-07-12 \

 宇通协议会把定位、车况和仪表里程拆成不同的稀疏帧。当天只有明确 `meter_speed=0` 的定位帧、但没有新的仪表总里程帧时，stat-writer 会用同一来源最近的历史总里程生成一条显式 `0 km` 记录，`quality_reason=stationary_carry_forward`；后续收到真实总里程后仍会正常覆盖增长。`vehicle_stat_samples_total{status="recovered_stationary"}` 用于观察该恢复路径。历史补录只能在轨迹核验确认为 `0 km` 后，临时为单次 `stats-backfill` 增加 `BACKFILL_STATIONARY_CARRY_FORWARD=true`；该开关默认关闭，禁止写入 timer，避免仅凭最新静止状态推断整日未行驶。

-三类协议都优先采用终端上报的累计里程字段。部分终端只有坐标、速度而累计里程稀疏或停止刷新时，stat-writer 会按同一 VIN、自然日、协议和物理来源累计相邻坐标距离，状态持久化在 `vehicle_daily_gps_mileage_state`，并以 `quality_reason=gps_coordinate_accumulation` 写入低优先级候选。该路径既覆盖总里程字段缺失，也覆盖字段存在但数值与上一帧相同；重复总里程仍计入 `skipped_same_mileage` 证据，只有同时具备有效移动点时才增加 GPS 候选。JT808 兼容没有速度字段的存量终端；GB32960、宇通 MQTT 必须同时满足设备速度大于 0，且实时状态写入按 10 秒节流，避免静止 GPS 漂移和 1 秒级宇通上报放大 MySQL 压力。相邻点超过 10 分钟、时间倒序或隐含速度超过 220 km/h 的线段不累计；低于 100 米的自然日累计只保留在状态表中，不投影为业务候选，避免瞬时车速噪声和微小定位漂移生成虚假日里程。选举顺序为“正向累计里程 > 正向 GPS 轨迹估算 > 累计里程零值占位”，因此稀疏累计里程造成的错误 0 km 会被已有移动轨迹纠正，真实正向累计里程恢复后仍会自动覆盖估算值；禁止把估算值混入平台累计里程归并。`vehicle_stat_samples_total{status="recovered_gps_coordinate"}` 和 `vehicle_stat_cache_entries{cache="gps_accumulation"}` 用于观察实时恢复量及节流缓存。`stats-backfill` 默认用 TDengine `vehicle_locations` 为缺少正向累计里程但存在可用移动线段的车辆补建同口径候选；可临时设置 `BACKFILL_GPS_FALLBACK=false` 禁用全部协议的降级路径，旧环境变量 `BACKFILL_JT808_GPS_FALLBACK` 继续兼容。
+三类协议都优先采用终端上报的累计里程字段。部分终端只有坐标、速度而累计里程稀疏或停止刷新时，stat-writer 会按同一 VIN、自然日、协议和物理来源累计相邻坐标距离，状态持久化在 `vehicle_daily_gps_mileage_state`，并以 `quality_reason=gps_coordinate_accumulation` 写入低优先级候选。该路径既覆盖总里程字段缺失，也覆盖字段存在但数值与上一帧相同；重复总里程仍计入 `skipped_same_mileage` 证据，只有同时具备有效移动点时才增加 GPS 候选。JT808 兼容没有速度字段的存量终端；GB32960、宇通 MQTT 必须同时满足设备速度大于 0，且实时状态写入按 10 秒节流，避免静止 GPS 漂移和 1 秒级宇通上报放大 MySQL 压力。相邻点超过 10 分钟、时间倒序或隐含速度超过 220 km/h 的线段不累计；低于 100 米的自然日累计只保留在状态表中，不投影为业务候选，避免瞬时车速噪声和微小定位漂移生成虚假日里程。`stats-backfill` 每次扫描还会查找窗口内阈值规则上线前遗留的低于 100 米 GPS 候选：dry-run 只报告数量，正式回补在单车/日期/协议事务中删除候选并重新执行来源选举，确保最终表不会继续保留旧漂移值。选举顺序为“正向累计里程 > 正向 GPS 轨迹估算 > 累计里程零值占位”，因此稀疏累计里程造成的错误 0 km 会被已有移动轨迹纠正，真实正向累计里程恢复后仍会自动覆盖估算值；禁止把估算值混入平台累计里程归并。`vehicle_stat_samples_total{status="recovered_gps_coordinate"}` 和 `vehicle_stat_cache_entries{cache="gps_accumulation"}` 用于观察实时恢复量及节流缓存。`stats-backfill` 默认用 TDengine `vehicle_locations` 为缺少正向累计里程但存在可用移动线段的车辆补建同口径候选；可临时设置 `BACKFILL_GPS_FALLBACK=false` 禁用全部协议的降级路径，旧环境变量 `BACKFILL_JT808_GPS_FALLBACK` 继续兼容。

 生产定时补算使用 systemd timer，默认每天 01:30 补算昨天往前 3 天，覆盖上游晚到或断传后恢复的总里程字段：

--- a/go/vehicle-gateway/cmd/stats-backfill/main.go
+++ b/go/vehicle-gateway/cmd/stats-backfill/main.go
@@ -155,6 +155,10 @@ func main() {
 		if err != nil {
 			fail("build GPS-coordinate fallback aggregates", err)
 		}
+		subthresholdFound, subthresholdDeleted, err := cleanupSubthresholdGPSMileageCandidates(ctx, mysqlDB, cfg, !cfg.DryRun)
+		if err != nil {
+			fail("cleanup sub-threshold GPS mileage candidates", err)
+		}
 		var written int64
 		var normalized int
 		if !cfg.DryRun {
@@ -167,7 +171,7 @@ func main() {
 				fail("normalize historical platform sources", err)
 			}
 		}
-		slog.Info("stats backfill complete", "method", cfg.Method, "dry_run", cfg.DryRun, "aggregates", len(aggregates), "realtimeLocationFallbacks", fallbacks, "jt808GPSFallbacks", gpsFallbacks, "written", written, "platformSourcesNormalized", normalized)
+		slog.Info("stats backfill complete", "method", cfg.Method, "dry_run", cfg.DryRun, "aggregates", len(aggregates), "realtimeLocationFallbacks", fallbacks, "jt808GPSFallbacks", gpsFallbacks, "subthresholdGPSCandidatesFound", subthresholdFound, "subthresholdGPSCandidatesDeleted", subthresholdDeleted, "written", written, "platformSourcesNormalized", normalized)
 		return
 	}

@@ -516,6 +520,100 @@ type gpsCoordinateBackfillState struct {
 	longGaps        int64
 }

+type subthresholdGPSMileageTarget struct {
+	vin      string
+	statDate string
+	protocol envelope.Protocol
+	rowCount int64
+}
+
+func cleanupSubthresholdGPSMileageCandidates(ctx context.Context, db *sql.DB, cfg config, apply bool) (int64, int64, error) {
+	if db == nil || len(cfg.Protocols) == 0 {
+		return 0, 0, nil
+	}
+	placeholders := make([]string, 0, len(cfg.Protocols))
+	args := []any{
+		cfg.DateFrom,
+		cfg.DateTo,
+		stats.QualityReasonGPSCoordinate,
+		stats.GPSMinimumDailyDistanceKM,
+	}
+	for _, protocol := range cfg.Protocols {
+		placeholders = append(placeholders, "?")
+		args = append(args, string(protocol))
+	}
+	query := `SELECT vin, DATE_FORMAT(stat_date, '%Y-%m-%d'), protocol, COUNT(*)
+FROM vehicle_daily_mileage_source
+WHERE stat_date >= ? AND stat_date <= ?
+  AND quality_reason = ?
+  AND daily_mileage_km >= 0 AND daily_mileage_km < ?
+  AND protocol IN (` + strings.Join(placeholders, ",") + `)
+GROUP BY vin, stat_date, protocol
+ORDER BY stat_date, protocol, vin`
+	rows, err := db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return 0, 0, err
+	}
+	targets := make([]subthresholdGPSMileageTarget, 0)
+	var found int64
+	for rows.Next() {
+		var target subthresholdGPSMileageTarget
+		var protocol string
+		if err := rows.Scan(&target.vin, &target.statDate, &protocol, &target.rowCount); err != nil {
+			_ = rows.Close()
+			return found, 0, err
+		}
+		target.protocol = envelope.Protocol(strings.TrimSpace(protocol))
+		targets = append(targets, target)
+		found += target.rowCount
+	}
+	if err := rows.Err(); err != nil {
+		_ = rows.Close()
+		return found, 0, err
+	}
+	if err := rows.Close(); err != nil {
+		return found, 0, err
+	}
+	if !apply {
+		return found, 0, nil
+	}
+
+	var deleted int64
+	for _, target := range targets {
+		tx, err := db.BeginTx(ctx, nil)
+		if err != nil {
+			return found, deleted, err
+		}
+		result, err := tx.ExecContext(ctx, `DELETE FROM vehicle_daily_mileage_source
+WHERE vin = ? AND stat_date = ? AND protocol = ?
+  AND quality_reason = ?
+  AND daily_mileage_km >= 0 AND daily_mileage_km < ?`,
+			target.vin,
+			target.statDate,
+			string(target.protocol),
+			stats.QualityReasonGPSCoordinate,
+			stats.GPSMinimumDailyDistanceKM,
+		)
+		if err == nil {
+			err = stats.ProjectDailyMileage(ctx, tx, target.vin, target.statDate, target.protocol)
+		}
+		if err != nil {
+			_ = tx.Rollback()
+			return found, deleted, err
+		}
+		affected, err := result.RowsAffected()
+		if err != nil {
+			_ = tx.Rollback()
+			return found, deleted, err
+		}
+		if err := tx.Commit(); err != nil {
+			return found, deleted, err
+		}
+		deleted += affected
+	}
+	return found, deleted, nil
+}
+
 func addGPSCoordinateFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, tdDB *sql.DB, cfg config, aggregates map[string]*metricAgg) (int, error) {
 	if !cfg.GPSFallback || tdDB == nil || aggregates == nil {
 		return 0, nil
--- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go
+++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go
@@ -50,6 +50,73 @@ func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T)
 	}
 }

+func TestCleanupSubthresholdGPSMileageCandidatesDryRunReportsWithoutDeleting(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	mock.ExpectQuery("SELECT vin, DATE_FORMAT").
+		WithArgs("2026-07-20", "2026-07-20", stats.QualityReasonGPSCoordinate, stats.GPSMinimumDailyDistanceKM, "JT808").
+		WillReturnRows(sqlmock.NewRows([]string{"vin", "stat_date", "protocol", "row_count"}).
+			AddRow("LKLG7C4E9NA774739", "2026-07-20", "JT808", 1))
+
+	found, deleted, err := cleanupSubthresholdGPSMileageCandidates(context.Background(), db, config{
+		DateFrom:  "2026-07-20",
+		DateTo:    "2026-07-20",
+		Protocols: []envelope.Protocol{envelope.ProtocolJT808},
+	}, false)
+	if err != nil {
+		t.Fatalf("cleanupSubthresholdGPSMileageCandidates() error = %v", err)
+	}
+	if found != 1 || deleted != 0 {
+		t.Fatalf("cleanup result found=%d deleted=%d", found, deleted)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet SQL expectations: %v", err)
+	}
+}
+
+func TestCleanupSubthresholdGPSMileageCandidatesDeletesAndReprojectsAtomically(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	mock.ExpectQuery("SELECT vin, DATE_FORMAT").
+		WithArgs("2026-07-20", "2026-07-20", stats.QualityReasonGPSCoordinate, stats.GPSMinimumDailyDistanceKM, "JT808").
+		WillReturnRows(sqlmock.NewRows([]string{"vin", "stat_date", "protocol", "row_count"}).
+			AddRow("LKLG7C4E9NA774739", "2026-07-20", "JT808", 1))
+	mock.ExpectBegin()
+	mock.ExpectExec("DELETE FROM vehicle_daily_mileage_source").
+		WithArgs("LKLG7C4E9NA774739", "2026-07-20", "JT808", stats.QualityReasonGPSCoordinate, stats.GPSMinimumDailyDistanceKM).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec("INSERT INTO vehicle_daily_mileage").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec("UPDATE vehicle_daily_mileage_source").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec("DELETE FROM vehicle_daily_mileage").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectCommit()
+
+	found, deleted, err := cleanupSubthresholdGPSMileageCandidates(context.Background(), db, config{
+		DateFrom:  "2026-07-20",
+		DateTo:    "2026-07-20",
+		Protocols: []envelope.Protocol{envelope.ProtocolJT808},
+	}, true)
+	if err != nil {
+		t.Fatalf("cleanupSubthresholdGPSMileageCandidates() error = %v", err)
+	}
+	if found != 1 || deleted != 1 {
+		t.Fatalf("cleanup result found=%d deleted=%d", found, deleted)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("unmet SQL expectations: %v", err)
+	}
+}
+
 func TestChooseTrustedSourceAcceptsSmallNegativeMileageJitter(t *testing.T) {
 	previous := []dailySourceLast{{
 		VIN:            "LB9A32A28R0LS1574",
--- a/go/vehicle-gateway/internal/stats/gps_mileage.go
+++ b/go/vehicle-gateway/internal/stats/gps_mileage.go
@@ -17,7 +17,7 @@ const (
 	gpsCoordinateSourceSuffix  = "#GPS_COORDINATE"
 	gpsMaxSegmentGap           = 10 * time.Minute
 	gpsMaxImpliedSpeedKMH      = 220.0
-	gpsMinimumDailyDistanceKM  = 0.1
+	GPSMinimumDailyDistanceKM  = 0.1
 )

 const GPSMileageStateTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_gps_mileage_state (
@@ -266,7 +266,7 @@ func AccumulateGPSMileage(ctx context.Context, exec Execer, point GPSMileagePoin
 func GPSMileageCandidateEligible(distanceKM float64, usableSegmentCount int64) bool {
 	return usableSegmentCount > 0 &&
 		isFiniteNonNegative(distanceKM) &&
-		distanceKM >= gpsMinimumDailyDistanceKM
+		distanceKM >= GPSMinimumDailyDistanceKM
 }

 // AccumulateJT808GPSMileage remains as a compatibility wrapper for callers and
--- a/go/vehicle-gateway/internal/stats/gps_mileage_test.go
+++ b/go/vehicle-gateway/internal/stats/gps_mileage_test.go
@@ -299,7 +299,7 @@ func TestAccumulateGPSMileageKeepsSubThresholdDistanceAsStateOnly(t *testing.T)
 	if err != nil {
 		t.Fatalf("AccumulateGPSMileage() error = %v", err)
 	}
-	if recovered || state.DailyMileageKM <= 0 || state.DailyMileageKM >= gpsMinimumDailyDistanceKM {
+	if recovered || state.DailyMileageKM <= 0 || state.DailyMileageKM >= GPSMinimumDailyDistanceKM {
 		t.Fatalf("sub-threshold state = %+v recovered=%v", state, recovered)
 	}
 	if err := mock.ExpectationsWereMet(); err != nil {