功能：完善里程边界与氢耗流式统计

2026-09-02 14:05:04 +08:00
parent 4201878c35
commit 8759668d39
23 changed files with 1638 additions and 159 deletions
@@ -513,12 +513,22 @@ BACKFILL_DATE_TO=2026-07-12 \

 三类协议都优先采用终端上报的累计里程字段。部分终端只有坐标、速度而累计里程稀疏或停止刷新时，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` 继续兼容。

-GB32960 与宇通 MQTT 的每日里程同时产出 `pure_hydrogen_mileage_km`。GB32960 采用燃料电池扩展中的 `engine_work_state=2`，宇通采用 `TRIANGLE_STATE IN (4,11)` 判定燃料电池处于纯氢工作模式；只有相邻两个有效累计里程点均为已识别的纯氢模式时才累计该段距离，模式切换、未知值、倒序和异常跳变均不计入纯氢里程。候选来源证据保存在 `vehicle_daily_mileage_source`，最终值随每日里程来源选举投影到 `vehicle_daily_mileage`，并始终限制为不大于对应日总里程。
+GB32960 与宇通 MQTT 的每日里程同时产出 `pure_hydrogen_mileage_km`。GB32960 使用整车运行模式：`0x01` 为纯电、`0x02` 为混动、`0x03` 为燃油、`0xFE` 为异常、`0xFF` 为无效；氢能车辆以当前帧运行模式归属相邻里程区间，模式 `0x02` 计入纯氢里程、模式 `0x01` 计入纯电里程，因此在只有有效纯电/混动模式的车辆日满足“纯氢里程＝总里程－纯电模式里程”。燃油、异常、无效、倒序和异常跳变不计入纯氢里程并应作为质量异常检查。宇通继续采用 `TRIANGLE_STATE IN (4,11)` 判定氢能参与。候选来源证据保存在 `vehicle_daily_mileage_source`，最终值随每日里程来源选举投影到 `vehicle_daily_mileage`，并始终限制为不大于对应日总里程。燃料电池启停仍由燃料电池子系统工作状态/电流判定，不与整车运行模式混用。

 内部里程统计会关联 `vehicle_open_daily_energy` 中 `quality_status=OK` 的日用氢量，展示 `耗氢 kg`，并用同一车辆日的 `耗氢量 ÷ 纯氢里程 × 100` 计算百公里纯氢耗。区间百公里纯氢耗只汇总同时具备有效氢耗和纯氢里程的车辆日；`NO_DATA`、`SUSPECT` 或没有纯氢里程的数据展示为 `—`，禁止当作 `0` 或混入分母。

 存量日期需要用原始报文扫描方式回算纯氢里程，因为默认 `last_diff` 只读取每日首尾累计里程，无法还原日内工作模式序列。先保持 `BACKFILL_DRY_RUN=true` 核对扫描量，再以 `BACKFILL_METHOD=scan BACKFILL_DRY_RUN=false BACKFILL_PROTOCOLS=GB32960,YUTONG_MQTT` 对指定日期窗口执行正式回算；扫描按事件时间顺序读取原始报文，并使用与实时统计完全相同的模式判定和相邻里程规则。

+全车队重放必须使用 `scripts/replay-pure-hydrogen-mileage.sh`。脚本先取氢瓶容量配置 VIN 与目标窗口内真实存在 GB32960 帧的 VIN 交集，再按单 VIN 查询并受控并发，避免全表 `ORDER BY event_time`。默认只 dry-run；确认后显式增加 `--apply`。每台车保留独立日志，失败可按 VIN 重跑：
+
+```bash
+/opt/lingniu-go-native/tools/replay-pure-hydrogen-mileage.sh \
+  --date-from 2026-08-01 \
+  --date-to 2026-08-18 \
+  --parallel 4 \
+  --apply
+```
+
 生产定时补算使用 systemd timer，默认每天 01:30 补算昨天往前 3 天，覆盖上游晚到或断传后恢复的总里程字段：

 ```bash
@@ -174,6 +174,15 @@ func syncHydrogenCapacities(ctx context.Context, logger interface {
 		logger.Warn("hydrogen tank capacity sync failed; retaining current cache", "source_schema", cfg.HydrogenCapacitySourceSchema, "error", err)
 		return
 	}
+	energyResult, energyErr := stats.SyncHydrogenEnergyParameters(syncCtx, db, cfg.HydrogenCapacitySourceSchema)
+	if energyErr != nil {
+		registry.IncCounter("vehicle_stat_hydrogen_energy_parameter_sync_total", metrics.Labels{"status": "error"})
+		logger.Warn("hydrogen battery capacity sync failed", "source_schema", cfg.HydrogenCapacitySourceSchema, "error", energyErr)
+	} else {
+		registry.IncCounter("vehicle_stat_hydrogen_energy_parameter_sync_total", metrics.Labels{"status": "ok"})
+		metrics.RecordLastActivity(registry, "vehicle_stat_hydrogen_energy_parameter_last_sync_unix_seconds", nil)
+		logger.Info("hydrogen battery capacities synchronized", "source_schema", cfg.HydrogenCapacitySourceSchema, "read", energyResult.Read, "written", energyResult.Written, "deactivated", energyResult.Deactivated)
+	}
 	loaded, err := writer.ReloadHydrogenTankCapacities(syncCtx)
 	if err != nil {
 		registry.IncCounter("vehicle_stat_hydrogen_capacity_sync_total", metrics.Labels{"status": "cache_error"})
@@ -29,6 +29,7 @@ type config struct {
 	DateFrom          string
 	DateTo            string
 	Protocols         []envelope.Protocol
+	VINs              []string
 	Method            string
 	Limit             int
 	DryRun            bool
@@ -39,6 +40,8 @@ type config struct {
 	GPSFallback       bool
 	ProgressEvery     int64
 	BaselineLookback  int
+	ListHydrogenVINs  bool
+	ListActiveVINs    bool
 	Location          *time.Location
 }

@@ -117,6 +120,21 @@ func main() {
 	}
 	ctx := context.Background()

+	mysqlDB, err := sql.Open("mysql", cfg.MySQLDSN)
+	if err != nil {
+		fail("open mysql", err)
+	}
+	defer mysqlDB.Close()
+	if err := mysqlDB.PingContext(ctx); err != nil {
+		fail("ping mysql", err)
+	}
+	if cfg.ListHydrogenVINs {
+		if err := printActiveHydrogenVINs(ctx, mysqlDB); err != nil {
+			fail("list active hydrogen vins", err)
+		}
+		return
+	}
+
 	td, err := sql.Open(cfg.TDengineDriver, cfg.TDengineDSN)
 	if err != nil {
 		fail("open tdengine", err)
@@ -125,14 +143,11 @@ func main() {
 	if err := td.PingContext(ctx); err != nil {
 		fail("ping tdengine", err)
 	}
-
-	mysqlDB, err := sql.Open("mysql", cfg.MySQLDSN)
-	if err != nil {
-		fail("open mysql", err)
+	if cfg.ListActiveVINs {
+		if err := printActiveFrameVINs(ctx, td, cfg); err != nil {
+			fail("list active frame vins", err)
 		}
-	defer mysqlDB.Close()
-	if err := mysqlDB.PingContext(ctx); err != nil {
-		fail("ping mysql", err)
+		return
 	}

 	schemaWriter := stats.NewWriter(mysqlDB, cfg.Location)
@@ -516,10 +531,11 @@ func addRealtimeLocationFallbackAggregates(ctx context.Context, mysqlDB *sql.DB,
 						currentRow.TotalKM = previousRow.TotalKM
 						currentSources[i] = currentRow
 					}
+					usesPreviousDayBoundary := hasPrevious && stats.IsUsableDailyMileageBoundary(date, previousRow.TS, currentRow.FirstTS.Location())
 					agg := aggregateFromDailySource(date, protocol, currentRow, previousRow, hasPrevious)
 					if cfg.StationaryCarry {
 						agg.QualityReason = stats.QualityReasonStationaryCarry
-					} else if hasPrevious {
+					} else if usesPreviousDayBoundary {
 						agg.QualityReason = "realtime_location_fallback_historical_baseline"
 					} else {
 						agg.QualityReason = "realtime_location_fallback_current_day_first_baseline"
@@ -909,7 +925,7 @@ func aggregateFromDailySource(date string, protocol envelope.Protocol, current d
 	firstEventTime := current.FirstTS
 	qualityStatus := stats.QualityOK
 	qualityReason := stats.QualityReasonCurrentDayFirst
-	if hasPrevious {
+	if hasPrevious && stats.IsUsableDailyMileageBoundary(date, previous.TS, current.FirstTS.Location()) {
 		firstKM = previous.TotalKM
 		firstEventTime = previous.TS
 		qualityStatus = stats.QualityOK
@@ -1295,10 +1311,32 @@ func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[str
 		}
 		fields[key] = value
 	}
+	promoteVehicleRunningModeForStats(protocol, fields, text)
 	promoteFuelCellWorkModeForStats(protocol, fields, text)
 	return fields
 }

+func promoteVehicleRunningModeForStats(protocol envelope.Protocol, fields map[string]any, text string) {
+	if protocol != envelope.ProtocolGB32960 || fields == nil {
+		return
+	}
+	if _, exists := fields[envelope.FieldVehicleRunningMode]; exists {
+		return
+	}
+	for _, key := range []string{"gb32960.vehicle.running_mode", "running_mode"} {
+		if value, exists := fields[key]; exists {
+			fields[envelope.FieldVehicleRunningMode] = value
+			return
+		}
+	}
+	for _, key := range []string{"gb32960.vehicle.running_mode", "running_mode"} {
+		if value, ok := extractJSONStringField(text, key); ok {
+			fields[envelope.FieldVehicleRunningMode] = value
+			return
+		}
+	}
+}
+
 func promoteFuelCellWorkModeForStats(protocol envelope.Protocol, fields map[string]any, text string) {
 	if fields == nil {
 		return
@@ -1383,6 +1421,7 @@ func loadConfig() (config, error) {
 		DateFrom:          dateFrom,
 		DateTo:            dateTo,
 		Protocols:         protocols,
+		VINs:              parseVINs(env("BACKFILL_VINS", "")),
 		Method:            env("BACKFILL_METHOD", "last_diff"),
 		Limit:             envInt("BACKFILL_LIMIT", 0),
 		DryRun:            envBool("BACKFILL_DRY_RUN", true),
@@ -1393,10 +1432,29 @@ func loadConfig() (config, error) {
 		GPSFallback:       envBool("BACKFILL_GPS_FALLBACK", envBool("BACKFILL_JT808_GPS_FALLBACK", true)),
 		ProgressEvery:     int64(envInt("BACKFILL_PROGRESS_EVERY", 100000)),
 		BaselineLookback:  envInt("BACKFILL_BASELINE_LOOKBACK_DAYS", 7),
+		ListHydrogenVINs:  envBool("BACKFILL_LIST_HYDROGEN_VINS", false),
+		ListActiveVINs:    envBool("BACKFILL_LIST_ACTIVE_VINS", false),
 		Location:          loc,
 	}, nil
 }

+func parseVINs(value string) []string {
+	seen := make(map[string]struct{})
+	vins := make([]string, 0)
+	for _, raw := range strings.Split(value, ",") {
+		vin := strings.ToUpper(strings.TrimSpace(raw))
+		if vin == "" {
+			continue
+		}
+		if _, ok := seen[vin]; ok {
+			continue
+		}
+		seen[vin] = struct{}{}
+		vins = append(vins, vin)
+	}
+	return vins
+}
+
 func resolveBackfillDateRange(now time.Time, loc *time.Location) (string, string) {
 	if loc == nil {
 		loc = time.FixedZone("Asia/Shanghai", 8*3600)
@@ -1441,6 +1499,13 @@ func queryRawFrames(ctx context.Context, db *sql.DB, cfg config) (*sql.Rows, err
 		}
 		where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")")
 	}
+	if len(cfg.VINs) > 0 {
+		quoted := make([]string, 0, len(cfg.VINs))
+		for _, vin := range cfg.VINs {
+			quoted = append(quoted, "'"+quote(vin)+"'")
+		}
+		where = append(where, "vin IN ("+strings.Join(quoted, ",")+")")
+	}
 	where = append(where, realtimeMileageFramePredicate())
 	sqlText := fmt.Sprintf(`SELECT protocol, vin, phone, device_id, source_endpoint, event_id, message_id, event_time, received_at, parsed_json, raw_text
 FROM %s.raw_frames
@@ -1449,10 +1514,61 @@ WHERE %s
 	if cfg.Limit > 0 {
 		sqlText += fmt.Sprintf(" LIMIT %d", cfg.Limit)
 	}
-	slog.Info("query raw frames", "date_from", cfg.DateFrom, "date_to", cfg.DateTo, "protocols", cfg.Protocols, "limit", cfg.Limit)
+	slog.Info("query raw frames", "date_from", cfg.DateFrom, "date_to", cfg.DateTo, "protocols", cfg.Protocols, "vins", cfg.VINs, "limit", cfg.Limit)
 	return db.QueryContext(ctx, sqlText)
 }

+func printActiveHydrogenVINs(ctx context.Context, db *sql.DB) error {
+	rows, err := db.QueryContext(ctx, `SELECT DISTINCT UPPER(TRIM(vin))
+FROM vehicle_hydrogen_tank_capacity
+WHERE active = 1 AND tank_capacity_l > 0 AND vin IS NOT NULL AND TRIM(vin) <> ''
+ORDER BY UPPER(TRIM(vin))`)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var vin string
+		if err := rows.Scan(&vin); err != nil {
+			return err
+		}
+		fmt.Println(vin)
+	}
+	return rows.Err()
+}
+
+func printActiveFrameVINs(ctx context.Context, db *sql.DB, cfg config) error {
+	where := backfillTimePredicates(cfg, cfg.DateFrom, nextDate(cfg.DateTo))
+	where = append(where,
+		"parse_status = 'OK'",
+		"vin IS NOT NULL",
+		"vin <> ''",
+	)
+	if len(cfg.Protocols) > 0 {
+		quoted := make([]string, 0, len(cfg.Protocols))
+		for _, protocol := range cfg.Protocols {
+			quoted = append(quoted, "'"+quote(string(protocol))+"'")
+		}
+		where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")")
+	}
+	where = append(where, realtimeMileageFramePredicate())
+	rows, err := db.QueryContext(ctx, fmt.Sprintf(`SELECT DISTINCT vin
+FROM %s.raw_frames
+WHERE %s`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")))
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var vin string
+		if err := rows.Scan(&vin); err != nil {
+			return err
+		}
+		fmt.Println(strings.ToUpper(strings.TrimSpace(vin)))
+	}
+	return rows.Err()
+}
+
 func backfillTimePredicates(cfg config, eventDateFrom string, eventDateToExclusive string) []string {
 	where := make([]string, 0, 4)
 	if !cfg.EventTimeFullScan {
@@ -4,6 +4,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
@@ -189,7 +190,7 @@ func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) {
 	}
 }

-func TestAggregateFromDailySourceUsesOlderHistoricalBaseline(t *testing.T) {
+func TestAggregateFromDailySourceUsesCurrentDayBoundaryAfterOfflineGap(t *testing.T) {
 	current := dailySourceLast{
 		VIN:            "LMRKH9AC2R1004087",
 		SourceKey:      normalizedSourceKey("YUTONG_MQTT", "", "LMRKH9AC2R1004087", "mqtt://yutong/ytforward/shln/3"),
@@ -212,13 +213,13 @@ func TestAggregateFromDailySourceUsesOlderHistoricalBaseline(t *testing.T) {

 	agg := aggregateFromDailySource("2026-07-08", envelope.ProtocolYutongMQTT, current, previous, true)

-	if agg.FirstKM != 120672 || agg.LatestKM != 120788 {
+	if agg.FirstKM != 120778 || agg.LatestKM != 120788 {
 		t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM)
 	}
-	if agg.FirstEventTime != previous.TS || agg.LatestEventTime != current.TS {
+	if agg.FirstEventTime != current.FirstTS || agg.LatestEventTime != current.TS {
 		t.Fatalf("event range = %v -> %v", agg.FirstEventTime, agg.LatestEventTime)
 	}
-	if agg.QualityStatus != stats.QualityOK || agg.QualityReason != stats.QualityReasonHistorical {
+	if agg.QualityStatus != stats.QualityOK || agg.QualityReason != stats.QualityReasonCurrentDayFirst {
 		t.Fatalf("quality = %s/%s", agg.QualityStatus, agg.QualityReason)
 	}
 	if agg.Count != 15 {
@@ -226,7 +227,7 @@ func TestAggregateFromDailySourceUsesOlderHistoricalBaseline(t *testing.T) {
 	}
 }

-func TestAggregateFromDailySourceRejectsHistoricalBaselineJump(t *testing.T) {
+func TestAggregateFromDailySourceIgnoresHistoricalBaselineJumpAcrossOfflineGap(t *testing.T) {
 	loc := time.FixedZone("Asia/Shanghai", 8*3600)
 	current := dailySourceLast{
 		VIN:            "LNXNEGRR6SR319464",
@@ -248,10 +249,10 @@ func TestAggregateFromDailySourceRejectsHistoricalBaselineJump(t *testing.T) {

 	agg := aggregateFromDailySource("2026-07-12", envelope.ProtocolGB32960, current, previous, true)

-	if agg.FirstKM != previous.TotalKM || agg.LatestKM != current.TotalKM {
+	if agg.FirstKM != current.FirstTotalKM || agg.LatestKM != current.TotalKM {
 		t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM)
 	}
-	if !agg.FirstEventTime.Equal(previous.TS) || !agg.LatestEventTime.Equal(current.TS) {
+	if !agg.FirstEventTime.Equal(current.FirstTS) || !agg.LatestEventTime.Equal(current.TS) {
 		t.Fatalf("event range = %v -> %v", agg.FirstEventTime, agg.LatestEventTime)
 	}
 	if agg.QualityStatus != stats.QualityInvalidDelta || agg.QualityReason != "outside_daily_range" {
@@ -285,7 +286,7 @@ func TestAggregateFromDailySourceUsesCurrentDayFirstWhenHistoryIsEmpty(t *testin
 	}
 }

-func TestBuildLastDiffAggregatesCarriesNearestHistoryAcrossEmptyDays(t *testing.T) {
+func TestBuildLastDiffAggregatesStartsFreshBoundaryAfterEmptyDays(t *testing.T) {
 	tdDB, mock, err := sqlmock.New()
 	if err != nil {
 		t.Fatalf("sqlmock.New() error = %v", err)
@@ -333,11 +334,11 @@ func TestBuildLastDiffAggregatesCarriesNearestHistoryAcrossEmptyDays(t *testing.
 	if agg == nil {
 		t.Fatalf("missing day-three aggregate; keys=%v", aggregateKeys(aggregates))
 	}
-	if agg.FirstKM != 100 || agg.LatestKM != 120 {
-		t.Fatalf("km range = %v -> %v, want 100 -> 120", agg.FirstKM, agg.LatestKM)
+	if agg.FirstKM != 120 || agg.LatestKM != 120 {
+		t.Fatalf("km range = %v -> %v, want 120 -> 120", agg.FirstKM, agg.LatestKM)
 	}
-	if !agg.FirstEventTime.Equal(dayOneTS) || agg.QualityReason != stats.QualityReasonHistorical {
-		t.Fatalf("baseline = %v reason=%q, want day-one historical baseline", agg.FirstEventTime, agg.QualityReason)
+	if !agg.FirstEventTime.Equal(dayThreeTS) || agg.QualityReason != stats.QualityReasonCurrentDayFirst {
+		t.Fatalf("baseline = %v reason=%q, want day-three current-day baseline", agg.FirstEventTime, agg.QualityReason)
 	}
 	if err := mock.ExpectationsWereMet(); err != nil {
 		t.Fatalf("sql expectations: %v", err)
@@ -601,10 +602,10 @@ func TestAddRealtimeLocationFallbackReusesAggregateHistoryAcrossEmptyDays(t *tes
 	if agg == nil {
 		t.Fatal("missing realtime-location fallback aggregate")
 	}
-	if agg.FirstKM != 100 || agg.LatestKM != 120 || !agg.FirstEventTime.Equal(dayOneTS) {
-		t.Fatalf("fallback range = %v@%v -> %v, want 100@day-one -> 120", agg.FirstKM, agg.FirstEventTime, agg.LatestKM)
+	if agg.FirstKM != 120 || agg.LatestKM != 120 || !agg.FirstEventTime.Equal(dayThreeTS) {
+		t.Fatalf("fallback range = %v@%v -> %v, want 120@day-three -> 120", agg.FirstKM, agg.FirstEventTime, agg.LatestKM)
 	}
-	if agg.QualityReason != "realtime_location_fallback_historical_baseline" {
+	if agg.QualityReason != "realtime_location_fallback_current_day_first_baseline" {
 		t.Fatalf("quality reason = %q", agg.QualityReason)
 	}
 	if err := mysqlMock.ExpectationsWereMet(); err != nil {
@@ -823,6 +824,24 @@ func TestFieldsForStatsPromotesGB32960FuelCellEngineWorkState(t *testing.T) {
 	}
 }

+func TestFieldsForStatsPromotesGB32960VehicleRunningMode(t *testing.T) {
+	fields := fieldsForStats(envelope.ProtocolGB32960, "LTEST000000000001", `{
+	  "gb32960.vehicle.total_mileage_km": "120.5",
+	  "gb32960.vehicle.running_mode": "1"
+	}`)
+
+	if got := fields[envelope.FieldVehicleRunningMode]; got == nil {
+		t.Fatalf("fields missing GB32960 vehicle running mode: %#v", fields)
+	}
+	active, known := stats.PureHydrogenModeFromEnvelope(envelope.FrameEnvelope{
+		Protocol: envelope.ProtocolGB32960,
+		Fields:   fields,
+	})
+	if !known || active {
+		t.Fatalf("pure-hydrogen mode = known:%v active:%v, want known inactive", known, active)
+	}
+}
+
 func TestClearBackfillTargetMileageClearsExactKey(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	if err != nil {
@@ -1028,7 +1047,7 @@ func TestAddSamplesUsesCurrentDayFirstSampleBaseline(t *testing.T) {
 	}
 }

-func TestAddSamplesBackfillsOnlyContinuousPureHydrogenIntervals(t *testing.T) {
+func TestAddSamplesBackfillsTotalMileageMinusPureElectricIntervals(t *testing.T) {
 	loc := time.FixedZone("Asia/Shanghai", 8*3600)
 	base := stats.MetricSample{
 		VIN:                   "LTEST000000000001",
@@ -1061,8 +1080,8 @@ func TestAddSamplesBackfillsOnlyContinuousPureHydrogenIntervals(t *testing.T) {
 		t.Fatalf("aggregate count = %d, want 1", len(aggregates))
 	}
 	for _, agg := range aggregates {
-		if agg.PureHydrogenMileageKM != 5 {
-			t.Fatalf("pure-hydrogen mileage = %v, want 5", agg.PureHydrogenMileageKM)
+		if agg.PureHydrogenMileageKM != 7 {
+			t.Fatalf("pure-hydrogen mileage = %v, want 7", agg.PureHydrogenMileageKM)
 		}
 		if agg.PureHydrogenSampleCount != 2 {
 			t.Fatalf("pure-hydrogen sample count = %d, want 2", agg.PureHydrogenSampleCount)
@@ -1094,6 +1113,14 @@ func TestLoadConfigDefaultsBackfillMethodToLastDiff(t *testing.T) {
 	}
 }

+func TestParseVINsNormalizesAndDeduplicates(t *testing.T) {
+	got := parseVINs(" la9gg68l1pbaf4814, LFV123 ,LA9GG68L1PBAF4814, ")
+	want := []string{"LA9GG68L1PBAF4814", "LFV123"}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("parseVINs() = %v, want %v", got, want)
+	}
+}
+
 func TestBackfillTimePredicatesUsePrimaryTimeForCoarseScanAndEventTimeForBusinessDay(t *testing.T) {
 	where := strings.Join(backfillTimePredicates(config{}, "2026-07-13", "2026-07-14"), " AND ")
 	for _, want := range []string{
@@ -38,6 +38,7 @@ const (
 	FieldLongitude          = "longitude"
 	FieldLatitude           = "latitude"
 	FieldSOCPercent         = "soc_percent"
+	FieldVehicleRunningMode = "running_mode"
 	FieldFuelCellWorkMode   = "fuel_cell_work_mode"
 )

@@ -145,7 +145,7 @@ func parseDataBody(version string, body []byte, fields map[string]any) (time.Tim
 			units = append(units, map[string]any{"type": "0x01", "name": "vehicle", "value": unit})
 			fields["vehicle_status"] = unit["vehicle_status"]
 			fields["charge_status"] = unit["charge_status"]
-			fields["running_mode"] = unit["running_mode"]
+			fields[envelope.FieldVehicleRunningMode] = unit["running_mode"]
 			fields[envelope.FieldSpeedKMH] = unit["speed_kmh"]
 			fields[envelope.FieldTotalMileageKM] = unit["total_mileage_km"]
 			fields[envelope.FieldSOCPercent] = unit["soc_percent"]
@@ -197,6 +197,7 @@ func TestParseFrameKeepsParsingRealFuelCellReportUntilUnknownExtension(t *testin
 		t.Fatalf("vin = %q", env.VIN)
 	}
 	assertFloatField(t, env, envelope.FieldTotalMileageKM, 53490.9)
+	assertIntField(t, env, envelope.FieldVehicleRunningMode, 1)
 	assertIntField(t, env, envelope.FieldFuelCellWorkMode, 2)
 	assertFloatField(t, env, "fuel_cell_hydrogen_consumption_kg_per_100km", 1.8)
 	assertFloatField(t, env, envelope.FieldLongitude, 120.800326)
@@ -302,7 +302,9 @@ func (w *Writer) EnsureSchema(ctx context.Context) error {
 		DailyMileageSourceTableSQL,
 		DailyMileageTableSQL,
 		HydrogenTankCapacityTableSQL,
+		HydrogenEnergyParameterTableSQL,
 		HydrogenStreamStateTableSQL,
+		HydrogenSegmentStreamStateTableSQL,
 	} {
 		if _, err := w.exec.ExecContext(ctx, statement); err != nil {
 			return err
@@ -433,6 +435,15 @@ func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelop
 			if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil {
 				return result, err
 			}
+			// A new natural-day/source row may start from the previous day's
+			// odometer baseline. Attribute that initial delta to the current
+			// running mode so pure-hydrogen mileage remains the complement of
+			// pure-electric mileage from the first accepted sample onward.
+			if candidate.QualityStatus == QualityOK &&
+				candidate.LatestPureHydrogenModeKnown &&
+				candidate.LatestPureHydrogenActive {
+				candidate.PureHydrogenMileageKM = candidate.DailyKM
+			}
 		}
 		projectDaily := w.shouldProjectDailyMileage(sample)
 		if projectDaily {
@@ -868,6 +879,16 @@ func (w *Writer) applyRealtimeBaseline(ctx context.Context, candidate *SourceMil
 		return err
 	}
 	if !found {
+		protocolBaseline, protocolFound, protocolErr := lookupPreviousProtocolBaseline(ctx, w.query, candidate.VIN, candidate.StatDate, candidate.Protocol)
+		if protocolErr != nil {
+			return protocolErr
+		}
+		if protocolFound && candidate.LatestTotalKM+maxNegativeMileageJitterKM < protocolBaseline.LatestTotalKM {
+			candidate.DailyKM = 0
+			candidate.QualityStatus = QualityInvalidDelta
+			candidate.QualityReason = QualityReasonTotalRollback
+			return nil
+		}
 		// If no earlier odometer exists at all, use the first current-day sample.
 		// Later samples retain that boundary through the in-memory baseline cache.
 		candidate.DailyKM = 0
@@ -875,9 +896,25 @@ func (w *Writer) applyRealtimeBaseline(ctx context.Context, candidate *SourceMil
 		candidate.QualityReason = QualityReasonCurrentDayFirst
 		return nil
 	}
+	if !IsUsableDailyMileageBoundary(candidate.StatDate, baseline.LatestEventTime, w.loc) {
+		// A vehicle recovering after one or more empty natural days must start a
+		// fresh day boundary. Otherwise the complete offline-period odometer
+		// increase is incorrectly assigned to the recovery day.
+		candidate.FirstTotalKM = candidate.LatestTotalKM
+		candidate.FirstEventTime = candidate.LatestEventTime
+		candidate.DailyKM = 0
+		candidate.QualityStatus = QualityOK
+		candidate.QualityReason = QualityReasonCurrentDayFirst
+		return nil
+	}
 	candidate.FirstTotalKM = baseline.LatestTotalKM
 	candidate.FirstEventTime = baseline.LatestEventTime
 	candidate.DailyKM = DailyMileageFromDayBoundary(baseline.LatestTotalKM, candidate.LatestTotalKM)
+	if candidate.LatestTotalKM+maxNegativeMileageJitterKM < baseline.LatestTotalKM {
+		candidate.QualityStatus = QualityInvalidDelta
+		candidate.QualityReason = QualityReasonTotalRollback
+		return nil
+	}
 	candidate.QualityStatus = QualityOK
 	candidate.QualityReason = baseline.QualityReason
 	if candidate.QualityReason == "" {
@@ -1054,6 +1054,9 @@ func TestWriterAppendUsesCurrentSampleWhenNoHistoricalBaseline(t *testing.T) {
 	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
 		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135").
 		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
 	mock.ExpectBegin()
 	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
 		WithArgs(
@@ -1138,6 +1141,9 @@ func TestWriterCachesMissingBaselineAfterSuccessfulFirstSample(t *testing.T) {
 	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
 		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135").
 		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
 	mock.ExpectBegin()
 	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
 		WithArgs(
@@ -1447,7 +1453,7 @@ func TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(t *testing.T
 	}
 }

-func TestWriterAppendUsesOlderHistoricalSourceBaseline(t *testing.T) {
+func TestWriterAppendStartsCurrentDayBoundaryAfterOfflineGap(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	if err != nil {
 		t.Fatalf("sqlmock.New() error = %v", err)
@@ -1488,18 +1494,18 @@ func TestWriterAppendUsesOlderHistoricalSourceBaseline(t *testing.T) {
 			"",
 			"LMRKH9AC2R1004087",
 			"",
-			float64(120672.0),
 			float64(120788.0),
-			approxFloat64{want: 116.0, tolerance: 0.000001},
+			float64(120788.0),
+			float64(0),
 			float64(0),
 			int64(1),
 			int64(0),
-			historicalTime,
+			currentTime,
 			currentTime,
 			false,
 			false,
 			QualityOK,
-			QualityReasonHistorical,
+			QualityReasonCurrentDayFirst,
 		).
 		WillReturnResult(sqlmock.NewResult(0, 1))
 	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
@@ -1608,7 +1614,7 @@ func TestWriterApplyRealtimeBaselineKeepsNegativeHistoricalDeltaInvalid(t *testi
 	if err := writer.applyRealtimeBaseline(context.Background(), &candidate); err != nil {
 		t.Fatalf("applyRealtimeBaseline() error = %v", err)
 	}
-	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != "outside_daily_range" {
+	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != QualityReasonTotalRollback {
 		t.Fatalf("quality = %s/%s", candidate.QualityStatus, candidate.QualityReason)
 	}
 	if candidate.DailyKM > -55 || candidate.DailyKM < -56 {
@@ -1651,7 +1657,7 @@ func TestWriterApplyRealtimeBaselineKeepsInvalidWhenCurrentDayBaselineIsStillHig
 	if err := writer.applyRealtimeBaseline(context.Background(), &candidate); err != nil {
 		t.Fatalf("applyRealtimeBaseline() error = %v", err)
 	}
-	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != "outside_daily_range" {
+	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != QualityReasonTotalRollback {
 		t.Fatalf("quality = %s/%s", candidate.QualityStatus, candidate.QualityReason)
 	}
 	if candidate.DailyKM > -55 || candidate.DailyKM < -56 {
@@ -1691,7 +1697,7 @@ func TestWriterApplyRealtimeBaselineRejectsNegativeDeltaEvenWithLowerCurrentDayS
 	if err := writer.applyRealtimeBaseline(context.Background(), &candidate); err != nil {
 		t.Fatalf("applyRealtimeBaseline() error = %v", err)
 	}
-	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != "outside_daily_range" {
+	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != QualityReasonTotalRollback {
 		t.Fatalf("quality = %s/%s", candidate.QualityStatus, candidate.QualityReason)
 	}
 	if candidate.DailyKM > -55 || candidate.DailyKM < -56 {
@@ -1705,7 +1711,7 @@ func TestWriterApplyRealtimeBaselineRejectsNegativeDeltaEvenWithLowerCurrentDayS
 	}
 }

-func TestWriterApplyRealtimeBaselineAcceptsPlausibleMultiDayFallbackDelta(t *testing.T) {
+func TestWriterApplyRealtimeBaselineIgnoresPlausibleMultiDayFallbackDelta(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	if err != nil {
 		t.Fatalf("sqlmock.New() error = %v", err)
@@ -1734,18 +1740,65 @@ func TestWriterApplyRealtimeBaselineAcceptsPlausibleMultiDayFallbackDelta(t *tes
 	if err := writer.applyRealtimeBaseline(context.Background(), &candidate); err != nil {
 		t.Fatalf("applyRealtimeBaseline() error = %v", err)
 	}
-	if candidate.QualityStatus != QualityOK || candidate.QualityReason != QualityReasonHistorical {
+	if candidate.QualityStatus != QualityOK || candidate.QualityReason != QualityReasonCurrentDayFirst {
 		t.Fatalf("quality = %s/%s", candidate.QualityStatus, candidate.QualityReason)
 	}
-	if candidate.DailyKM < 7833.4 || candidate.DailyKM > 7833.6 {
-		t.Fatalf("daily km = %v, want multi-day gap delta", candidate.DailyKM)
+	if candidate.DailyKM != 0 {
+		t.Fatalf("daily km = %v, want current-day first baseline after offline gap", candidate.DailyKM)
+	}
+	if candidate.FirstTotalKM != candidate.LatestTotalKM || !candidate.FirstEventTime.Equal(currentTime) {
+		t.Fatalf("current-day boundary not retained: %#v", candidate)
 	}
 	if err := mock.ExpectationsWereMet(); err != nil {
 		t.Fatalf("sql expectations: %v", err)
 	}
 }

-func TestWriterApplyRealtimeBaselineRejectsHistoricalBaselineJump(t *testing.T) {
+func TestWriterApplyRealtimeBaselineRejectsProtocolOdometerRollbackAfterSourceChange(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	writer := NewWriter(db, loc)
+	previousTime := time.Date(2026, 8, 5, 22, 41, 0, 0, loc)
+	currentTime := time.Date(2026, 8, 6, 8, 12, 0, 0, loc)
+	candidate := SourceMileageSample{
+		VIN:             "LNXNEGRR1SR321395",
+		StatDate:        "2026-08-06",
+		Protocol:        envelope.ProtocolGB32960,
+		SourceKey:       "GB32960:new-device@8.134.95.166",
+		LatestTotalKM:   0.1,
+		LatestEventTime: currentTime,
+	}
+
+	// The new device has no source-key baseline, but this VIN already has a
+	// durable GB32960 odometer. The reset must not become a NORMAL day start.
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LNXNEGRR1SR321395", "2026-08-06", "GB32960", "GB32960:new-device@8.134.95.166").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LNXNEGRR1SR321395", "2026-08-06", "GB32960").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}).
+			AddRow(48233.3, previousTime))
+
+	if err := writer.applyRealtimeBaseline(context.Background(), &candidate); err != nil {
+		t.Fatalf("applyRealtimeBaseline() error = %v", err)
+	}
+	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != QualityReasonTotalRollback {
+		t.Fatalf("quality = %s/%s, want %s/%s", candidate.QualityStatus, candidate.QualityReason, QualityInvalidDelta, QualityReasonTotalRollback)
+	}
+	if candidate.DailyKM != 0 {
+		t.Fatalf("daily km = %v, want 0", candidate.DailyKM)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestWriterApplyRealtimeBaselineIgnoresHistoricalBaselineJumpAcrossOfflineGap(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	if err != nil {
 		t.Fatalf("sqlmock.New() error = %v", err)
@@ -1774,14 +1827,14 @@ func TestWriterApplyRealtimeBaselineRejectsHistoricalBaselineJump(t *testing.T)
 	if err := writer.applyRealtimeBaseline(context.Background(), &candidate); err != nil {
 		t.Fatalf("applyRealtimeBaseline() error = %v", err)
 	}
-	if candidate.QualityStatus != QualityInvalidDelta || candidate.QualityReason != "outside_daily_range" {
+	if candidate.QualityStatus != QualityOK || candidate.QualityReason != QualityReasonCurrentDayFirst {
 		t.Fatalf("quality = %s/%s", candidate.QualityStatus, candidate.QualityReason)
 	}
-	if candidate.DailyKM < 29999.9 || candidate.DailyKM > 30000.1 {
-		t.Fatalf("daily km = %v, want historical jump delta", candidate.DailyKM)
+	if candidate.DailyKM != 0 {
+		t.Fatalf("daily km = %v, want current-day first baseline", candidate.DailyKM)
 	}
-	if candidate.FirstTotalKM != 10009.7 || !candidate.FirstEventTime.Equal(baselineTime) {
-		t.Fatalf("previous-day baseline not preserved: %#v", candidate)
+	if candidate.FirstTotalKM != candidate.LatestTotalKM || !candidate.FirstEventTime.Equal(currentTime) {
+		t.Fatalf("current-day boundary not retained: %#v", candidate)
 	}
 	if err := mock.ExpectationsWereMet(); err != nil {
 		t.Fatalf("sql expectations: %v", err)
@@ -1815,6 +1868,9 @@ func TestWriterAppendMarksCandidateNoPreviousBaselineWhenPreviousBaselineMissing
 	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
 		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135").
 		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
 	mock.ExpectBegin()
 	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
 		WithArgs(
@@ -1894,7 +1950,7 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
 		t.Fatalf("Append() error = %v", err)
 	}

-	schemaCalls := 6 + len(DailyMileageAlterSQL)
+	schemaCalls := 8 + len(DailyMileageAlterSQL)
 	if len(exec.calls) != schemaCalls+5 {
 		t.Fatalf("exec calls = %d", len(exec.calls))
 	}
@@ -1904,7 +1960,9 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
 		"CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_source",
 		"CREATE TABLE IF NOT EXISTS vehicle_daily_mileage",
 		"CREATE TABLE IF NOT EXISTS vehicle_hydrogen_tank_capacity",
+		"CREATE TABLE IF NOT EXISTS vehicle_hydrogen_energy_parameter",
 		"CREATE TABLE IF NOT EXISTS vehicle_open_hydrogen_stream_state",
+		"CREATE TABLE IF NOT EXISTS vehicle_open_hydrogen_segment_stream_state",
 	} {
 		if !strings.Contains(exec.calls[i].query, want) {
 			t.Fatalf("schema call %d = %s, want %s", i, exec.calls[i].query, want)
@@ -2294,6 +2352,9 @@ func TestWriterRollsBackMileageTransactionWhenProjectionFails(t *testing.T) {
 	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
 		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135").
 		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
 	mock.ExpectBegin()
 	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
 		WithArgs(
@@ -0,0 +1,142 @@
+package stats
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"sort"
+	"strings"
+)
+
+const HydrogenEnergyParameterTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_hydrogen_energy_parameter (
+  vin VARCHAR(64) NOT NULL,
+  battery_capacity_kwh DECIMAL(12,3) NOT NULL,
+  hydrogen_energy_kwh_per_kg DECIMAL(12,3) NOT NULL DEFAULT 16.000,
+  confirmed_by VARCHAR(96) NOT NULL DEFAULT '',
+  source_note VARCHAR(255) NOT NULL DEFAULT '',
+  effective_from DATE NOT NULL,
+  effective_to DATE NULL,
+  active TINYINT(1) NOT NULL DEFAULT 1,
+  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+  PRIMARY KEY (vin, effective_from),
+  KEY idx_hydrogen_energy_parameter_active (active, effective_from, effective_to, vin),
+  CONSTRAINT chk_hydrogen_energy_parameter_capacity CHECK (battery_capacity_kwh > 0 AND battery_capacity_kwh <= 500),
+  CONSTRAINT chk_hydrogen_energy_parameter_conversion CHECK (hydrogen_energy_kwh_per_kg > 0 AND hydrogen_energy_kwh_per_kg <= 50),
+  CONSTRAINT chk_hydrogen_energy_parameter_dates CHECK (effective_to IS NULL OR effective_to >= effective_from)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`
+
+const (
+	hydrogenEnergyAssetSourceNote  = "ln_asset_management.vehicle_model.battery_capacity"
+	hydrogenEnergyAssetConfirmedBy = "资产车型主数据自动同步"
+	hydrogenEnergyAssetEffectiveAt = "1970-01-01"
+)
+
+type HydrogenEnergyParameterSyncResult struct {
+	Read        int
+	Written     int
+	Deactivated int64
+}
+
+type hydrogenEnergyParameterRow struct {
+	VIN             string
+	SourceVehicleID int64
+	CapacityKWh     float64
+	SourceUpdatedAt sql.NullTime
+}
+
+// SyncHydrogenEnergyParameters projects each active hydrogen vehicle's rated
+// battery capacity from the asset model master into the VIN-scoped calculation
+// parameter table. The fixed, oldest effective date lets a later effective-dated
+// business override win without disabling automatic master-data synchronization.
+func SyncHydrogenEnergyParameters(ctx context.Context, db *sql.DB, sourceSchema string) (HydrogenEnergyParameterSyncResult, error) {
+	result := HydrogenEnergyParameterSyncResult{}
+	if db == nil {
+		return result, fmt.Errorf("mysql database is required")
+	}
+	sourceSchema = strings.TrimSpace(sourceSchema)
+	if !mysqlIdentifier.MatchString(sourceSchema) {
+		return result, fmt.Errorf("invalid asset schema %q", sourceSchema)
+	}
+	query := fmt.Sprintf(`SELECT vi.id,UPPER(TRIM(vi.vin)),vm.battery_capacity,
+CASE
+  WHEN vi.update_time IS NULL THEN vm.update_time
+  WHEN vm.update_time IS NULL THEN vi.update_time
+  ELSE GREATEST(vi.update_time,vm.update_time)
+END
+FROM %s.vehicle_info vi
+JOIN %s.vehicle_model vm ON vm.id=vi.vehicle_model_id
+WHERE COALESCE(vi.del_flag,'0')='0' AND COALESCE(vm.del_flag,'0')='0'
+  AND LENGTH(TRIM(vi.vin))=17
+  AND vm.tank_capacity>0
+  AND vm.battery_capacity>0 AND vm.battery_capacity<=500`, sourceSchema, sourceSchema)
+	rows, err := db.QueryContext(ctx, query)
+	if err != nil {
+		return result, fmt.Errorf("read asset hydrogen energy parameters: %w", err)
+	}
+	byVIN := map[string]hydrogenEnergyParameterRow{}
+	for rows.Next() {
+		var row hydrogenEnergyParameterRow
+		if err := rows.Scan(&row.SourceVehicleID, &row.VIN, &row.CapacityKWh, &row.SourceUpdatedAt); err != nil {
+			rows.Close()
+			return result, err
+		}
+		result.Read++
+		if row.CapacityKWh <= 0 || row.CapacityKWh > 500 {
+			continue
+		}
+		if previous, exists := byVIN[row.VIN]; !exists || newerEnergyParameterRow(row, previous) {
+			byVIN[row.VIN] = row
+		}
+	}
+	if err := rows.Close(); err != nil {
+		return result, err
+	}
+
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return result, err
+	}
+	defer tx.Rollback()
+	if _, err := tx.ExecContext(ctx, `UPDATE vehicle_hydrogen_energy_parameter
+SET active=0
+WHERE active=1 AND source_note=?`, hydrogenEnergyAssetSourceNote); err != nil {
+		return result, err
+	}
+	const upsert = `INSERT INTO vehicle_hydrogen_energy_parameter(
+vin,battery_capacity_kwh,hydrogen_energy_kwh_per_kg,confirmed_by,source_note,effective_from,effective_to,active
+) VALUES(?,?,16.000,?,?,?,NULL,1)
+ON DUPLICATE KEY UPDATE battery_capacity_kwh=VALUES(battery_capacity_kwh),
+confirmed_by=VALUES(confirmed_by),source_note=VALUES(source_note),effective_to=NULL,active=1`
+	vins := make([]string, 0, len(byVIN))
+	for vin := range byVIN {
+		vins = append(vins, vin)
+	}
+	sort.Strings(vins)
+	for _, vin := range vins {
+		row := byVIN[vin]
+		if _, err := tx.ExecContext(ctx, upsert, row.VIN, row.CapacityKWh, hydrogenEnergyAssetConfirmedBy, hydrogenEnergyAssetSourceNote, hydrogenEnergyAssetEffectiveAt); err != nil {
+			return result, err
+		}
+		result.Written++
+	}
+	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*)
+FROM vehicle_hydrogen_energy_parameter
+WHERE active=0 AND source_note=?`, hydrogenEnergyAssetSourceNote).Scan(&result.Deactivated); err != nil {
+		return result, err
+	}
+	if err := tx.Commit(); err != nil {
+		return result, err
+	}
+	return result, nil
+}
+
+func newerEnergyParameterRow(left, right hydrogenEnergyParameterRow) bool {
+	if left.SourceUpdatedAt.Valid != right.SourceUpdatedAt.Valid {
+		return left.SourceUpdatedAt.Valid
+	}
+	if left.SourceUpdatedAt.Valid && !left.SourceUpdatedAt.Time.Equal(right.SourceUpdatedAt.Time) {
+		return left.SourceUpdatedAt.Time.After(right.SourceUpdatedAt.Time)
+	}
+	return left.SourceVehicleID > right.SourceVehicleID
+}
@@ -0,0 +1,95 @@
+package stats
+
+import (
+	"context"
+	"regexp"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+)
+
+func TestSyncHydrogenEnergyParametersCopiesAssetModelCapacity(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	updatedAt := time.Date(2026, 8, 26, 10, 0, 0, 0, time.Local)
+	mock.ExpectQuery("FROM ln_asset_management\\.vehicle_info vi").
+		WillReturnRows(sqlmock.NewRows([]string{"id", "vin", "battery_capacity", "updated_at"}).
+			AddRow(1001, "LB9A32A24R0LS1037", 21.04, updatedAt).
+			AddRow(1002, "LB9A52124P0LS1001", 58.73, updatedAt))
+	mock.ExpectBegin()
+	mock.ExpectExec(regexp.QuoteMeta("UPDATE vehicle_hydrogen_energy_parameter")).
+		WithArgs(hydrogenEnergyAssetSourceNote).
+		WillReturnResult(sqlmock.NewResult(0, 2))
+	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_hydrogen_energy_parameter(")).
+		WithArgs("LB9A32A24R0LS1037", 21.04, hydrogenEnergyAssetConfirmedBy, hydrogenEnergyAssetSourceNote, hydrogenEnergyAssetEffectiveAt).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_hydrogen_energy_parameter(")).
+		WithArgs("LB9A52124P0LS1001", 58.73, hydrogenEnergyAssetConfirmedBy, hydrogenEnergyAssetSourceNote, hydrogenEnergyAssetEffectiveAt).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectQuery(regexp.QuoteMeta("SELECT COUNT(*)")).
+		WithArgs(hydrogenEnergyAssetSourceNote).
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
+	mock.ExpectCommit()
+
+	result, err := SyncHydrogenEnergyParameters(context.Background(), db, "ln_asset_management")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.Read != 2 || result.Written != 2 || result.Deactivated != 0 {
+		t.Fatalf("result=%+v", result)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSyncHydrogenEnergyParametersKeepsNewestDuplicateVIN(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	oldTime := time.Date(2026, 8, 20, 10, 0, 0, 0, time.Local)
+	newTime := oldTime.Add(24 * time.Hour)
+	mock.ExpectQuery("FROM ln_asset_management\\.vehicle_info vi").
+		WillReturnRows(sqlmock.NewRows([]string{"id", "vin", "battery_capacity", "updated_at"}).
+			AddRow(1001, "LB9A32A24R0LS1037", 20.00, oldTime).
+			AddRow(1002, "LB9A32A24R0LS1037", 21.04, newTime))
+	mock.ExpectBegin()
+	mock.ExpectExec(regexp.QuoteMeta("UPDATE vehicle_hydrogen_energy_parameter")).
+		WithArgs(hydrogenEnergyAssetSourceNote).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_hydrogen_energy_parameter(")).
+		WithArgs("LB9A32A24R0LS1037", 21.04, hydrogenEnergyAssetConfirmedBy, hydrogenEnergyAssetSourceNote, hydrogenEnergyAssetEffectiveAt).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectQuery(regexp.QuoteMeta("SELECT COUNT(*)")).
+		WithArgs(hydrogenEnergyAssetSourceNote).
+		WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0))
+	mock.ExpectCommit()
+
+	result, err := SyncHydrogenEnergyParameters(context.Background(), db, "ln_asset_management")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.Read != 2 || result.Written != 1 {
+		t.Fatalf("result=%+v", result)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSyncHydrogenEnergyParametersRejectsUnsafeSchema(t *testing.T) {
+	db, _, err := sqlmock.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := SyncHydrogenEnergyParameters(context.Background(), db, "ln_asset_management;DROP"); err == nil {
+		t.Fatal("unsafe schema accepted")
+	}
+}
@@ -3,7 +3,10 @@ package stats
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
+	"errors"
 	"math"
+	"sort"
 	"strings"
 	"time"

@@ -14,8 +17,14 @@ import (
 const (
 	defaultHydrogenNoiseKG    = 0.05
 	defaultHydrogenMaxDropKG  = 20.0
+	hydrogenActiveCurrentA    = 1.0
 	hydrogenMolarMassKGPerMol = 0.00201588
 	universalGasConstant      = 8.314472
+	hydrogenSegmentWindowSize = 5
+	hydrogenSegmentMinSamples = hydrogenSegmentWindowSize * 2
+	hydrogenSegmentMaxGap     = 5 * time.Minute
+	hydrogenRecoveryWindow    = time.Minute
+	hydrogenRecoveryPressure  = 8.0
 )

 var hydrogenPressureFieldKeys = []string{
@@ -28,6 +37,11 @@ var hydrogenTemperatureFieldKeys = []string{
 	"fuel_cell_max_hydrogen_temperature_c",
 }

+var hydrogenCurrentFieldKeys = []string{
+	"gb32960.fuel_cell.fuel_cell_current_a",
+	"fuel_cell_current_a",
+}
+
 var hydrogenDensityA = [...]float64{0.05888460, -0.06136111, -0.002650473, 0.002731125, 0.001802374, -0.001150707, 0.00009588528, -0.0000001109040, 0.0000000001264403}
 var hydrogenDensityB = [...]float64{1.325, 1.87, 2.5, 2.8, 2.938, 3.14, 3.37, 3.75, 4.0}
 var hydrogenDensityC = [...]float64{1, 1, 2, 2, 2.42, 2.63, 3, 4, 5}
@@ -52,6 +66,8 @@ const HydrogenStreamStateTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_open_hyd
  first_event_time DATETIME(3) NOT NULL,
  last_event_time DATETIME(3) NOT NULL,
  last_event_id VARCHAR(128) NOT NULL DEFAULT '',
+  last_fuel_cell_active TINYINT(1) NOT NULL DEFAULT 0,
+  last_fuel_cell_state_known TINYINT(1) NOT NULL DEFAULT 0,
  quality_status VARCHAR(24) NOT NULL DEFAULT 'NO_DATA',
  quality_reason VARCHAR(255) NOT NULL DEFAULT '有效压力质量样本不足2条',
  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
@@ -60,6 +76,31 @@ const HydrogenStreamStateTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_open_hyd
  KEY idx_hydrogen_stream_date_quality (stat_date, quality_status, vin)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`

+const HydrogenSegmentStreamStateTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_open_hydrogen_segment_stream_state (
+  vin VARCHAR(64) NOT NULL,
+  stat_date DATE NOT NULL,
+  source_endpoint VARCHAR(128) NOT NULL DEFAULT '',
+  finalized_consumption_kg DECIMAL(18,3) NOT NULL DEFAULT 0,
+  projected_consumption_kg DECIMAL(18,3) NOT NULL DEFAULT 0,
+  sample_count BIGINT UNSIGNED NOT NULL DEFAULT 0,
+  refuel_count INT UNSIGNED NOT NULL DEFAULT 0,
+  abnormal_drop_count INT UNSIGNED NOT NULL DEFAULT 0,
+  eligible_interval_count BIGINT UNSIGNED NOT NULL DEFAULT 0,
+  qualified_segment_count INT UNSIGNED NOT NULL DEFAULT 0,
+  last_mass_kg DECIMAL(18,3) NOT NULL DEFAULT 0,
+  last_event_time DATETIME(3) NOT NULL,
+  last_event_id VARCHAR(128) NOT NULL DEFAULT '',
+  state_json JSON NOT NULL,
+  calculation_method VARCHAR(48) NOT NULL DEFAULT 'PRESSURE_NIST_SEGMENT_MEDIAN_5',
+  quality_status VARCHAR(24) NOT NULL DEFAULT 'NO_DATA',
+  quality_reason VARCHAR(255) NOT NULL DEFAULT '有效工作段关键帧不足10条',
+  created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+  updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+  PRIMARY KEY (vin, stat_date),
+  KEY idx_hydrogen_segment_stream_date_quality (stat_date, quality_status, vin),
+  CONSTRAINT chk_hydrogen_segment_stream_quality CHECK (quality_status IN ('OK', 'NO_DATA', 'SUSPECT'))
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`
+
 type HydrogenStreamSample struct {
 	VIN                 string
 	Date                string
@@ -72,6 +113,9 @@ type HydrogenStreamSample struct {
 	TemperatureC        float64
 	NoiseKG             float64
 	RefuelThresholdKG   float64
+	FuelCellActive      bool
+	FuelCellStateKnown  bool
+	ConsumptionEligible bool
 }

 type HydrogenStreamResult struct {
@@ -82,6 +126,268 @@ type HydrogenStreamResult struct {
 	NotCurrent int
 }

+type hydrogenStreamEndpoint struct {
+	MassKG  float64 `json:"massKg"`
+	NoiseKG float64 `json:"noiseKg"`
+}
+
+type hydrogenStreamSegmentWindow struct {
+	Count              int64                    `json:"count"`
+	First              []hydrogenStreamEndpoint `json:"first"`
+	Tail               []hydrogenStreamEndpoint `json:"tail"`
+	CycleMinimumMassKG float64                  `json:"cycleMinimumMassKg"`
+}
+
+type hydrogenSegmentStreamState struct {
+	SourceEndpoint         string                      `json:"sourceEndpoint"`
+	FinalizedConsumptionKG float64                     `json:"finalizedConsumptionKg"`
+	SampleCount            int64                       `json:"sampleCount"`
+	RefuelCount            int64                       `json:"refuelCount"`
+	AbnormalDropCount      int64                       `json:"abnormalDropCount"`
+	EligibleIntervalCount  int64                       `json:"eligibleIntervalCount"`
+	QualifiedSegmentCount  int64                       `json:"qualifiedSegmentCount"`
+	FirstMassKG            float64                     `json:"firstMassKg"`
+	RemainingMassKG        float64                     `json:"remainingMassKg"`
+	MetadataCycleMinimumKG float64                     `json:"metadataCycleMinimumKg"`
+	LastObservedMassKG     float64                     `json:"lastObservedMassKg"`
+	TankCapacityLiters     float64                     `json:"tankCapacityLiters"`
+	FirstPressureMPa       float64                     `json:"firstPressureMpa"`
+	LastPressureMPa        float64                     `json:"lastPressureMpa"`
+	FirstTemperatureC      float64                     `json:"firstTemperatureC"`
+	LastTemperatureC       float64                     `json:"lastTemperatureC"`
+	FirstEventTime         time.Time                   `json:"firstEventTime"`
+	LastEventTime          time.Time                   `json:"lastEventTime"`
+	LastEventID            string                      `json:"lastEventId"`
+	LastFuelCellActive     bool                        `json:"lastFuelCellActive"`
+	LastFuelCellStateKnown bool                        `json:"lastFuelCellStateKnown"`
+	Segment                hydrogenStreamSegmentWindow `json:"segment"`
+}
+
+type hydrogenStreamBaseline struct {
+	SourceEndpoint string
+	ConsumptionKG  float64
+	FirstMassKG    float64
+	LastMassKG     float64
+	SampleCount    int64
+	RefuelCount    int64
+	QualityStatus  string
+}
+
+func newHydrogenSegmentStreamState(sample HydrogenStreamSample, baseline *hydrogenStreamBaseline) hydrogenSegmentStreamState {
+	state := hydrogenSegmentStreamState{
+		SourceEndpoint: sample.SourceEndpoint,
+		FirstMassKG:    sample.MassKG, RemainingMassKG: sample.MassKG,
+		MetadataCycleMinimumKG: sample.MassKG, LastObservedMassKG: sample.MassKG,
+		TankCapacityLiters: sample.TankCapacityLiters,
+		FirstPressureMPa:   sample.PressureMPa, LastPressureMPa: sample.PressureMPa,
+		FirstTemperatureC: sample.TemperatureC, LastTemperatureC: sample.TemperatureC,
+		FirstEventTime: sample.EventTime, LastEventTime: sample.EventTime, LastEventID: sample.EventID,
+		LastFuelCellActive: sample.FuelCellActive, LastFuelCellStateKnown: sample.FuelCellStateKnown,
+		SampleCount: 1,
+	}
+	if baseline != nil {
+		state.SourceEndpoint = firstNonEmptyHydrogen(sample.SourceEndpoint, baseline.SourceEndpoint)
+		state.FinalizedConsumptionKG = baseline.ConsumptionKG
+		state.SampleCount += baseline.SampleCount
+		state.RefuelCount = baseline.RefuelCount
+		if baseline.FirstMassKG >= 0 {
+			state.FirstMassKG = baseline.FirstMassKG
+		}
+		if baseline.LastMassKG >= 0 {
+			state.RemainingMassKG = baseline.LastMassKG
+		}
+		if baseline.QualityStatus == "OK" || baseline.QualityStatus == "SUSPECT" {
+			state.EligibleIntervalCount = 1
+			state.QualifiedSegmentCount = 1
+		}
+		if baseline.QualityStatus == "SUSPECT" {
+			state.AbnormalDropCount = 1
+		}
+	}
+	if hydrogenStreamCanStartSegment(sample) {
+		state.startSegment(sample)
+	}
+	return state
+}
+
+func (state *hydrogenSegmentStreamState) add(sample HydrogenStreamSample) bool {
+	if !sample.EventTime.After(state.LastEventTime) {
+		return false
+	}
+	previousTime := state.LastEventTime
+	previousMass := state.LastObservedMassKG
+	previousPressure := state.LastPressureMPa
+	intervalEligible := hydrogenStreamIntervalEligible(
+		state.LastFuelCellActive, state.LastFuelCellStateKnown,
+		sample.FuelCellActive, sample.FuelCellStateKnown,
+	)
+	if intervalEligible || (sample.FuelCellStateKnown && sample.FuelCellActive) ||
+		sample.MassKG-state.RemainingMassKG > sample.NoiseKG {
+		state.RemainingMassKG = sample.MassKG
+	}
+	delta := state.MetadataCycleMinimumKG - sample.MassKG
+	switch {
+	case sample.MassKG-state.MetadataCycleMinimumKG > sample.RefuelThresholdKG:
+		if !hydrogenStreamRapidRecovery(previousTime, previousPressure, sample) {
+			state.RefuelCount++
+		}
+		state.MetadataCycleMinimumKG = sample.MassKG
+	case delta > sample.NoiseKG && delta <= defaultHydrogenMaxDropKG:
+		state.MetadataCycleMinimumKG = sample.MassKG
+	}
+
+	if intervalEligible {
+		state.EligibleIntervalCount++
+	}
+	switch {
+	case state.Segment.Count == 0 || !intervalEligible:
+		state.flushSegment()
+		if hydrogenStreamCanStartSegment(sample) {
+			state.startSegment(sample)
+		}
+	case sample.EventTime.Sub(previousTime) > hydrogenSegmentMaxGap:
+		state.flushSegment()
+		state.startSegment(sample)
+	case previousMass-sample.MassKG > defaultHydrogenMaxDropKG:
+		state.AbnormalDropCount++
+		state.flushSegment()
+		state.startSegment(sample)
+	case sample.MassKG-state.Segment.CycleMinimumMassKG > sample.RefuelThresholdKG:
+		state.flushSegment()
+		state.startSegment(sample)
+	default:
+		state.appendSegmentSample(sample)
+		if sample.MassKG < state.Segment.CycleMinimumMassKG {
+			state.Segment.CycleMinimumMassKG = sample.MassKG
+		}
+	}
+
+	state.SourceEndpoint = sample.SourceEndpoint
+	state.SampleCount++
+	state.LastObservedMassKG = sample.MassKG
+	state.TankCapacityLiters = sample.TankCapacityLiters
+	state.LastPressureMPa = sample.PressureMPa
+	state.LastTemperatureC = sample.TemperatureC
+	state.LastEventTime = sample.EventTime
+	state.LastEventID = sample.EventID
+	state.LastFuelCellActive = sample.FuelCellActive
+	state.LastFuelCellStateKnown = sample.FuelCellStateKnown
+	return true
+}
+
+func (state *hydrogenSegmentStreamState) startSegment(sample HydrogenStreamSample) {
+	state.Segment = hydrogenStreamSegmentWindow{CycleMinimumMassKG: sample.MassKG}
+	state.appendSegmentSample(sample)
+}
+
+func (state *hydrogenSegmentStreamState) appendSegmentSample(sample HydrogenStreamSample) {
+	endpoint := hydrogenStreamEndpoint{MassKG: sample.MassKG, NoiseKG: sample.NoiseKG}
+	state.Segment.Count++
+	if len(state.Segment.First) < hydrogenSegmentWindowSize {
+		state.Segment.First = append(state.Segment.First, endpoint)
+	}
+	if len(state.Segment.Tail) == hydrogenSegmentWindowSize {
+		copy(state.Segment.Tail, state.Segment.Tail[1:])
+		state.Segment.Tail[len(state.Segment.Tail)-1] = endpoint
+		return
+	}
+	state.Segment.Tail = append(state.Segment.Tail, endpoint)
+}
+
+func (state *hydrogenSegmentStreamState) flushSegment() {
+	if state.Segment.Count >= hydrogenSegmentMinSamples {
+		state.QualifiedSegmentCount++
+		state.FinalizedConsumptionKG += hydrogenStreamEndpointDrop(state.Segment.First, state.Segment.Tail)
+	}
+	state.Segment = hydrogenStreamSegmentWindow{}
+}
+
+func (state hydrogenSegmentStreamState) projectedConsumptionKG() float64 {
+	consumption := state.FinalizedConsumptionKG
+	if state.Segment.Count >= hydrogenSegmentMinSamples {
+		consumption += hydrogenStreamEndpointDrop(state.Segment.First, state.Segment.Tail)
+	}
+	return roundHydrogenKG(consumption)
+}
+
+func (state hydrogenSegmentStreamState) quality() (string, string) {
+	qualified := state.QualifiedSegmentCount
+	if state.Segment.Count >= hydrogenSegmentMinSamples {
+		qualified++
+	}
+	switch {
+	case state.SampleCount < 2:
+		return "NO_DATA", "有效车载氢量样本不足2条"
+	case state.EligibleIntervalCount == 0:
+		return "NO_DATA", "无燃料电池工作状态下的有效压力区间"
+	case qualified == 0:
+		return "NO_DATA", "有效工作段关键帧不足10条"
+	case state.AbnormalDropCount > 0:
+		return "SUSPECT", "存在超过阈值的异常下降"
+	default:
+		return "OK", ""
+	}
+}
+
+func hydrogenStreamCanStartSegment(sample HydrogenStreamSample) bool {
+	return !sample.FuelCellStateKnown || sample.FuelCellActive
+}
+
+func hydrogenStreamIntervalEligible(previousActive, previousKnown, currentActive, currentKnown bool) bool {
+	if previousKnown || currentKnown {
+		return previousKnown && previousActive && currentKnown && currentActive
+	}
+	return true
+}
+
+func hydrogenStreamRapidRecovery(previousTime time.Time, previousPressure float64, sample HydrogenStreamSample) bool {
+	elapsed := sample.EventTime.Sub(previousTime)
+	return elapsed > 0 && elapsed <= hydrogenRecoveryWindow &&
+		sample.PressureMPa-previousPressure >= hydrogenRecoveryPressure
+}
+
+func hydrogenStreamEndpointDrop(first, tail []hydrogenStreamEndpoint) float64 {
+	if len(first) < hydrogenSegmentWindowSize || len(tail) < hydrogenSegmentWindowSize {
+		return 0
+	}
+	startMass := hydrogenStreamMedianMass(first)
+	endMass := hydrogenStreamMedianMass(tail)
+	noise := defaultHydrogenNoiseKG
+	for _, endpoint := range first {
+		noise = math.Max(noise, endpoint.NoiseKG)
+	}
+	for _, endpoint := range tail {
+		noise = math.Max(noise, endpoint.NoiseKG)
+	}
+	drop := startMass - endMass
+	if drop <= noise {
+		return 0
+	}
+	return drop
+}
+
+func hydrogenStreamMedianMass(values []hydrogenStreamEndpoint) float64 {
+	masses := make([]float64, len(values))
+	for index, value := range values {
+		masses[index] = value.MassKG
+	}
+	sort.Float64s(masses)
+	return masses[len(masses)/2]
+}
+
+func roundHydrogenKG(value float64) float64 {
+	return math.Round(value*1000) / 1000
+}
+
+func firstNonEmptyHydrogen(values ...string) string {
+	for _, value := range values {
+		if strings.TrimSpace(value) != "" {
+			return strings.TrimSpace(value)
+		}
+	}
+	return ""
+}
+
 func HydrogenDensityKGPerM3(pressureMPa, temperatureC float64) (float64, bool) {
 	temperatureK := temperatureC + 273.15
 	if pressureMPa < 0 || pressureMPa > 70 || temperatureK < 220 || temperatureK > 1000 {
@@ -126,6 +432,9 @@ func HydrogenStreamSampleFromEnvelope(env envelope.FrameEnvelope, loc *time.Loca
 	if !temperatureOK {
 		return HydrogenStreamSample{}, "missing_temperature", false
 	}
+	if !validHydrogenPressureTemperature(pressure, temperature) {
+		return HydrogenStreamSample{}, "invalid_pressure_temperature", false
+	}
 	if tankCapacityLiters <= 0 || tankCapacityLiters > 10000 {
 		return HydrogenStreamSample{}, "missing_tank_capacity", false
 	}
@@ -137,6 +446,13 @@ func HydrogenStreamSampleFromEnvelope(env envelope.FrameEnvelope, loc *time.Loca
 	noiseKG := math.Max(defaultHydrogenNoiseKG, mass-pressureStepMass)
 	noiseKG = math.Min(noiseKG, 1.0)
 	refuelThresholdKG := math.Max(1.0, mass*0.05)
+	fuelCellActive, fuelCellStateKnown := FuelCellActiveModeFromEnvelope(env)
+	if !fuelCellStateKnown {
+		if currentA, currentOK := firstHydrogenNumber(env.Fields, hydrogenCurrentFieldKeys); currentOK && currentA >= 0 {
+			fuelCellActive = currentA > hydrogenActiveCurrentA
+			fuelCellStateKnown = true
+		}
+	}
 	if loc == nil {
 		loc = time.FixedZone("Asia/Shanghai", 8*3600)
 	}
@@ -157,9 +473,17 @@ func HydrogenStreamSampleFromEnvelope(env envelope.FrameEnvelope, loc *time.Loca
 		EventID: env.StableEventID(), EventTime: eventTime, MassKG: mass,
 		TankCapacityLiters: tankCapacityLiters, PressureMPa: pressure, TemperatureC: temperature,
 		NoiseKG: noiseKG, RefuelThresholdKG: refuelThresholdKG,
+		FuelCellActive: fuelCellActive, FuelCellStateKnown: fuelCellStateKnown,
+		ConsumptionEligible: !fuelCellStateKnown || fuelCellActive,
 	}, "", true
 }

+func validHydrogenPressureTemperature(pressureMPa, temperatureC float64) bool {
+	return pressureMPa > 0 && pressureMPa <= 70 && temperatureC > -40 && temperatureC <= 726.85 &&
+		!math.IsNaN(pressureMPa) && !math.IsInf(pressureMPa, 0) &&
+		!math.IsNaN(temperatureC) && !math.IsInf(temperatureC, 0)
+}
+
 func firstHydrogenNumber(fields map[string]any, keys []string) (float64, bool) {
 	for _, key := range keys {
 		if value, ok := telemetry.Number(fields, key); ok {
@@ -193,23 +517,32 @@ func AppendHydrogenStream(ctx context.Context, exec Execer, env envelope.FrameEn
 		return result, err
 	}
 	defer tx.Rollback()
-	write, err := tx.ExecContext(ctx, upsertHydrogenStreamStateSQL,
-		sample.VIN, sample.Date, sample.SourceEndpoint, sample.MassKG, sample.MassKG, sample.MassKG,
-		sample.TankCapacityLiters, sample.PressureMPa, sample.PressureMPa,
-		sample.TemperatureC, sample.TemperatureC, sample.EventTime, sample.EventTime, sample.EventID,
-		sample.NoiseKG, defaultHydrogenMaxDropKG, sample.RefuelThresholdKG,
-		defaultHydrogenMaxDropKG, defaultHydrogenMaxDropKG, defaultHydrogenMaxDropKG,
-		sample.RefuelThresholdKG, sample.NoiseKG, defaultHydrogenMaxDropKG,
-	)
+	state, found, err := selectHydrogenSegmentStreamState(ctx, tx, sample.VIN, sample.Date)
 	if err != nil {
 		return result, err
 	}
-	affected, _ := write.RowsAffected()
-	if affected == 0 {
+	if found {
+		if !state.add(sample) {
 			result.Duplicate = 1
 			return result, tx.Commit()
 		}
-	if _, err := tx.ExecContext(ctx, projectHydrogenStreamDailySQL, sample.VIN, sample.Date); err != nil {
+		if err := updateHydrogenSegmentStreamState(ctx, tx, sample.VIN, sample.Date, state); err != nil {
+			return result, err
+		}
+	} else {
+		baseline, baselineFound, err := selectHydrogenStreamBaseline(ctx, tx, sample.VIN, sample.Date)
+		if err != nil {
+			return result, err
+		}
+		if !baselineFound {
+			baseline = nil
+		}
+		state = newHydrogenSegmentStreamState(sample, baseline)
+		if err := insertHydrogenSegmentStreamState(ctx, tx, sample.VIN, sample.Date, state); err != nil {
+			return result, err
+		}
+	}
+	if err := projectHydrogenSegmentStreamDaily(ctx, tx, sample.VIN, sample.Date, state); err != nil {
 		return result, err
 	}
 	if err := tx.Commit(); err != nil {
@@ -219,50 +552,114 @@ func AppendHydrogenStream(ctx context.Context, exec Execer, env envelope.FrameEn
 	return result, nil
 }

-const upsertHydrogenStreamStateSQL = `
-INSERT INTO vehicle_open_hydrogen_stream_state(
-  vin,stat_date,source_endpoint,first_mass_kg,last_mass_kg,cycle_min_mass_kg,consumption_kg,sample_count,
-  refuel_count,abnormal_drop_count,tank_capacity_l,first_pressure_mpa,last_pressure_mpa,
-  first_temperature_c,last_temperature_c,calculation_method,
-  first_event_time,last_event_time,last_event_id,quality_status,quality_reason
-) VALUES(?,?,?,?,?,?,0,1,0,0,?,?,?,?,?,'PRESSURE_NIST',?,?,?,'NO_DATA','有效压力质量样本不足2条')
-ON DUPLICATE KEY UPDATE
-  consumption_kg = consumption_kg + IF(VALUES(last_event_time)>last_event_time
-    AND cycle_min_mass_kg-VALUES(last_mass_kg)>? AND cycle_min_mass_kg-VALUES(last_mass_kg)<=?,
-    cycle_min_mass_kg-VALUES(last_mass_kg),0),
-  refuel_count = refuel_count + IF(VALUES(last_event_time)>last_event_time
-    AND VALUES(last_mass_kg)-cycle_min_mass_kg>?,1,0),
-  quality_status = IF(VALUES(last_event_time)<=last_event_time,quality_status,
-    IF(abnormal_drop_count+IF(cycle_min_mass_kg-VALUES(last_mass_kg)>?,1,0)>0,'SUSPECT','OK')),
-  quality_reason = IF(VALUES(last_event_time)<=last_event_time,quality_reason,
-    IF(abnormal_drop_count+IF(cycle_min_mass_kg-VALUES(last_mass_kg)>?,1,0)>0,'存在超过阈值的异常下降','')),
-  abnormal_drop_count = abnormal_drop_count + IF(VALUES(last_event_time)>last_event_time
-    AND cycle_min_mass_kg-VALUES(last_mass_kg)>?,1,0),
-  cycle_min_mass_kg = IF(VALUES(last_event_time)>last_event_time,
-    IF(VALUES(last_mass_kg)-cycle_min_mass_kg>?,VALUES(last_mass_kg),
-      IF(cycle_min_mass_kg-VALUES(last_mass_kg)>? AND cycle_min_mass_kg-VALUES(last_mass_kg)<=?,VALUES(last_mass_kg),cycle_min_mass_kg)),
-    cycle_min_mass_kg),
-  sample_count = sample_count + IF(VALUES(last_event_time)>last_event_time,1,0),
-  tank_capacity_l = IF(VALUES(last_event_time)>last_event_time,VALUES(tank_capacity_l),tank_capacity_l),
-  last_pressure_mpa = IF(VALUES(last_event_time)>last_event_time,VALUES(last_pressure_mpa),last_pressure_mpa),
-  last_temperature_c = IF(VALUES(last_event_time)>last_event_time,VALUES(last_temperature_c),last_temperature_c),
-  calculation_method = 'PRESSURE_NIST',
-  last_mass_kg = IF(VALUES(last_event_time)>last_event_time,VALUES(last_mass_kg),last_mass_kg),
-  last_event_id = IF(VALUES(last_event_time)>last_event_time,VALUES(last_event_id),last_event_id),
-  last_event_time = GREATEST(last_event_time,VALUES(last_event_time))`
+func selectHydrogenSegmentStreamState(ctx context.Context, tx *sql.Tx, vin, date string) (hydrogenSegmentStreamState, bool, error) {
+	var encoded []byte
+	err := tx.QueryRowContext(ctx, selectHydrogenSegmentStreamStateSQL, vin, date).Scan(&encoded)
+	if errors.Is(err, sql.ErrNoRows) {
+		return hydrogenSegmentStreamState{}, false, nil
+	}
+	if err != nil {
+		return hydrogenSegmentStreamState{}, false, err
+	}
+	var state hydrogenSegmentStreamState
+	if err := json.Unmarshal(encoded, &state); err != nil {
+		return hydrogenSegmentStreamState{}, false, err
+	}
+	return state, true, nil
+}

-const projectHydrogenStreamDailySQL = `
-INSERT INTO vehicle_open_daily_energy(
+func selectHydrogenStreamBaseline(ctx context.Context, tx *sql.Tx, vin, date string) (*hydrogenStreamBaseline, bool, error) {
+	var baseline hydrogenStreamBaseline
+	var firstMass, lastMass sql.NullFloat64
+	err := tx.QueryRowContext(ctx, selectHydrogenStreamBaselineSQL, vin, date).Scan(
+		&baseline.SourceEndpoint, &baseline.ConsumptionKG, &firstMass, &lastMass,
+		&baseline.SampleCount, &baseline.RefuelCount, &baseline.QualityStatus,
+	)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, false, nil
+	}
+	if err != nil {
+		return nil, false, err
+	}
+	baseline.FirstMassKG = -1
+	baseline.LastMassKG = -1
+	if firstMass.Valid {
+		baseline.FirstMassKG = firstMass.Float64
+	}
+	if lastMass.Valid {
+		baseline.LastMassKG = lastMass.Float64
+	}
+	return &baseline, true, nil
+}
+
+func insertHydrogenSegmentStreamState(ctx context.Context, tx *sql.Tx, vin, date string, state hydrogenSegmentStreamState) error {
+	encoded, err := json.Marshal(state)
+	if err != nil {
+		return err
+	}
+	qualityStatus, qualityReason := state.quality()
+	_, err = tx.ExecContext(ctx, insertHydrogenSegmentStreamStateSQL,
+		vin, date, state.SourceEndpoint, state.FinalizedConsumptionKG, state.projectedConsumptionKG(),
+		state.SampleCount, state.RefuelCount, state.AbnormalDropCount, state.EligibleIntervalCount,
+		state.QualifiedSegmentCount, state.RemainingMassKG, state.LastEventTime, state.LastEventID,
+		encoded, qualityStatus, qualityReason,
+	)
+	return err
+}
+
+func updateHydrogenSegmentStreamState(ctx context.Context, tx *sql.Tx, vin, date string, state hydrogenSegmentStreamState) error {
+	encoded, err := json.Marshal(state)
+	if err != nil {
+		return err
+	}
+	qualityStatus, qualityReason := state.quality()
+	_, err = tx.ExecContext(ctx, updateHydrogenSegmentStreamStateSQL,
+		state.SourceEndpoint, state.FinalizedConsumptionKG, state.projectedConsumptionKG(),
+		state.SampleCount, state.RefuelCount, state.AbnormalDropCount, state.EligibleIntervalCount,
+		state.QualifiedSegmentCount, state.RemainingMassKG, state.LastEventTime, state.LastEventID,
+		encoded, qualityStatus, qualityReason, vin, date,
+	)
+	return err
+}
+
+func projectHydrogenSegmentStreamDaily(ctx context.Context, tx *sql.Tx, vin, date string, state hydrogenSegmentStreamState) error {
+	qualityStatus, qualityReason := state.quality()
+	_, err := tx.ExecContext(ctx, projectHydrogenSegmentStreamDailySQL,
+		vin, date, state.SourceEndpoint, state.projectedConsumptionKG(),
+		state.FirstMassKG, state.RemainingMassKG, state.SampleCount, state.RefuelCount,
+		qualityStatus, qualityReason,
+	)
+	return err
+}
+
+const selectHydrogenSegmentStreamStateSQL = `SELECT state_json
+FROM vehicle_open_hydrogen_segment_stream_state
+WHERE vin=? AND stat_date=?
+FOR UPDATE`
+
+const selectHydrogenStreamBaselineSQL = `SELECT source_endpoint,consumption_kg,first_mass_kg,last_mass_kg,
+  sample_count,refuel_count,quality_status
+FROM vehicle_open_daily_energy
+WHERE vin=? AND stat_date=? AND energy_type='HYDROGEN'
+LIMIT 1`
+
+const insertHydrogenSegmentStreamStateSQL = `INSERT INTO vehicle_open_hydrogen_segment_stream_state(
+  vin,stat_date,source_endpoint,finalized_consumption_kg,projected_consumption_kg,
+  sample_count,refuel_count,abnormal_drop_count,eligible_interval_count,qualified_segment_count,
+  last_mass_kg,last_event_time,last_event_id,state_json,calculation_method,quality_status,quality_reason
+) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,'PRESSURE_NIST_SEGMENT_MEDIAN_5',?,?)`
+
+const updateHydrogenSegmentStreamStateSQL = `UPDATE vehicle_open_hydrogen_segment_stream_state
+SET source_endpoint=?,finalized_consumption_kg=?,projected_consumption_kg=?,sample_count=?,
+    refuel_count=?,abnormal_drop_count=?,eligible_interval_count=?,qualified_segment_count=?,
+    last_mass_kg=?,last_event_time=?,last_event_id=?,state_json=?,
+    calculation_method='PRESSURE_NIST_SEGMENT_MEDIAN_5',quality_status=?,quality_reason=?
+WHERE vin=? AND stat_date=?`
+
+const projectHydrogenSegmentStreamDailySQL = `INSERT INTO vehicle_open_daily_energy(
  vin,stat_date,energy_type,source_endpoint,consumption_kg,unit,first_mass_kg,last_mass_kg,
  sample_count,refuel_count,quality_status,quality_reason,calculated_at
-)
-SELECT vin,stat_date,'HYDROGEN',source_endpoint,consumption_kg,'kg',first_mass_kg,last_mass_kg,
-       sample_count,refuel_count,quality_status,quality_reason,NOW(3)
-FROM vehicle_open_hydrogen_stream_state
-WHERE vin=? AND stat_date=?
-ORDER BY CASE quality_status WHEN 'OK' THEN 0 WHEN 'SUSPECT' THEN 1 ELSE 2 END,
-         sample_count DESC,source_endpoint ASC
-LIMIT 1
+) VALUES(?,?,'HYDROGEN',?,?,'kg',?,?,?,?,?,?,NOW(3))
 ON DUPLICATE KEY UPDATE
  source_endpoint=VALUES(source_endpoint),consumption_kg=VALUES(consumption_kg),unit='kg',
  first_mass_kg=VALUES(first_mass_kg),last_mass_kg=VALUES(last_mass_kg),
@@ -2,8 +2,10 @@ package stats

 import (
 	"context"
+	"database/sql"
 	"math"
 	"regexp"
+	"strings"
 	"testing"
 	"time"

@@ -47,6 +49,29 @@ func TestHydrogenStreamSampleUsesPressureTemperatureAndCapacity(t *testing.T) {
 	}
 }

+func TestHydrogenStreamMarksInactiveFuelCellIneligible(t *testing.T) {
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	eventTime := time.Date(2026, 8, 15, 13, 31, 7, 0, loc)
+	env := pressureEnvelope(eventTime)
+	env.Fields[envelope.FieldFuelCellWorkMode] = 0
+	sample, reason, ok := HydrogenStreamSampleFromEnvelope(env, loc, eventTime, 520)
+	if !ok || reason != "" || !sample.FuelCellStateKnown || sample.FuelCellActive || sample.ConsumptionEligible {
+		t.Fatalf("inactive sample classification=%#v reason=%q ok=%v", sample, reason, ok)
+	}
+}
+
+func TestHydrogenStreamFallsBackToFuelCellCurrent(t *testing.T) {
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	eventTime := time.Date(2026, 8, 15, 13, 31, 7, 0, loc)
+	env := pressureEnvelope(eventTime)
+	delete(env.Fields, envelope.FieldFuelCellWorkMode)
+	env.Fields["fuel_cell_current_a"] = 0.2
+	sample, reason, ok := HydrogenStreamSampleFromEnvelope(env, loc, eventTime, 520)
+	if !ok || reason != "" || !sample.FuelCellStateKnown || sample.FuelCellActive || sample.ConsumptionEligible {
+		t.Fatalf("current fallback classification=%#v reason=%q ok=%v", sample, reason, ok)
+	}
+}
+
 func TestHydrogenStreamRejectsMissingCapacity(t *testing.T) {
 	loc := time.FixedZone("Asia/Shanghai", 8*3600)
 	eventTime := time.Date(2026, 7, 21, 10, 30, 0, 0, loc)
@@ -56,6 +81,20 @@ func TestHydrogenStreamRejectsMissingCapacity(t *testing.T) {
 	}
 }

+func TestHydrogenStreamRejectsInvalidPressureTemperaturePlaceholders(t *testing.T) {
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	eventTime := time.Date(2026, 8, 12, 8, 11, 16, 0, loc)
+	for _, fields := range []struct{ pressure, temperature float64 }{{0, 30}, {22.8, -40}} {
+		env := pressureEnvelope(eventTime)
+		env.Fields["gb32960.fuel_cell.max_hydrogen_pressure_mpa"] = fields.pressure
+		env.Fields["gb32960.fuel_cell.max_hydrogen_temperature_c"] = fields.temperature
+		_, reason, ok := HydrogenStreamSampleFromEnvelope(env, loc, eventTime, 520)
+		if ok || reason != "invalid_pressure_temperature" {
+			t.Fatalf("placeholder pressure=%v temperature=%v accepted: ok=%v reason=%q", fields.pressure, fields.temperature, ok, reason)
+		}
+	}
+}
+
 func TestNISTHydrogenDensityValidationPoint(t *testing.T) {
 	density, ok := HydrogenDensityKGPerM3(10, 26.85)
 	if !ok || math.Abs(density-7.625) > 0.01 {
@@ -63,6 +102,144 @@ func TestNISTHydrogenDensityValidationPoint(t *testing.T) {
 	}
 }

+func hydrogenSegmentTestSample(at time.Time, mass float64, source string, active bool) HydrogenStreamSample {
+	return HydrogenStreamSample{
+		VIN: "LA9GG64L0NBAF4175", Date: at.Format("2006-01-02"), SourceEndpoint: source,
+		EventID: at.Format(time.RFC3339Nano), EventTime: at, MassKG: mass,
+		TankCapacityLiters: 520, PressureMPa: mass, TemperatureC: 30,
+		NoiseKG: 0.05, RefuelThresholdKG: 1,
+		FuelCellActive: active, FuelCellStateKnown: true, ConsumptionEligible: active,
+	}
+}
+
+func TestHydrogenSegmentStreamUsesEndpointMediansAcrossSources(t *testing.T) {
+	base := time.Date(2026, 8, 19, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
+	state := newHydrogenSegmentStreamState(hydrogenSegmentTestSample(base, 10, "source-a", true), nil)
+	for index := 1; index < 10; index++ {
+		mass := 10.0
+		if index >= 5 {
+			mass = 9.6
+		}
+		source := "source-a"
+		if index >= 6 {
+			source = "source-b"
+		}
+		if !state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), mass, source, true)) {
+			t.Fatalf("sample %d rejected", index)
+		}
+	}
+	quality, reason := state.quality()
+	if state.projectedConsumptionKG() != 0.4 || quality != "OK" || reason != "" || state.SampleCount != 10 {
+		t.Fatalf("state=%#v consumption=%.3f quality=%s reason=%q", state, state.projectedConsumptionKG(), quality, reason)
+	}
+	if state.SourceEndpoint != "source-b" || len(state.Segment.First) != 5 || len(state.Segment.Tail) != 5 {
+		t.Fatalf("endpoint/window state=%#v", state)
+	}
+}
+
+func TestHydrogenSegmentStreamFinalizesOnInactiveFrame(t *testing.T) {
+	base := time.Now()
+	state := newHydrogenSegmentStreamState(hydrogenSegmentTestSample(base, 10, "source-a", true), nil)
+	for index := 1; index < 10; index++ {
+		mass := 10.0
+		if index >= 5 {
+			mass = 9.6
+		}
+		state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), mass, "source-a", true))
+	}
+	state.add(hydrogenSegmentTestSample(base.Add(10*time.Second), 9.5, "source-a", false))
+	if math.Abs(state.FinalizedConsumptionKG-0.4) > 0.0001 || state.Segment.Count != 0 || state.QualifiedSegmentCount != 1 {
+		t.Fatalf("state=%#v", state)
+	}
+}
+
+func TestHydrogenSegmentStreamIgnoresOutOfOrderAndKeepsBoundedWindow(t *testing.T) {
+	base := time.Now()
+	state := newHydrogenSegmentStreamState(hydrogenSegmentTestSample(base, 20, "source-a", true), nil)
+	if state.add(hydrogenSegmentTestSample(base, 19, "source-b", true)) {
+		t.Fatal("same-time duplicate was accepted")
+	}
+	for index := 1; index <= 10000; index++ {
+		state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), 20-float64(index)/20000, "source-b", true))
+	}
+	if len(state.Segment.First) != 5 || len(state.Segment.Tail) != 5 || state.SampleCount != 10001 {
+		t.Fatalf("first=%d tail=%d samples=%d", len(state.Segment.First), len(state.Segment.Tail), state.SampleCount)
+	}
+}
+
+func TestHydrogenSegmentStreamInactiveDropDoesNotConsume(t *testing.T) {
+	base := time.Now()
+	state := newHydrogenSegmentStreamState(hydrogenSegmentTestSample(base, 10, "source-a", false), nil)
+	for index := 1; index < 20; index++ {
+		state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), 10-float64(index)/10, "source-a", false))
+	}
+	quality, _ := state.quality()
+	if state.projectedConsumptionKG() != 0 || state.RemainingMassKG != 10 || quality != "NO_DATA" {
+		t.Fatalf("state=%#v quality=%s", state, quality)
+	}
+}
+
+func TestHydrogenSegmentStreamSplitsAndAccumulatesRefuelCycles(t *testing.T) {
+	base := time.Now()
+	state := newHydrogenSegmentStreamState(hydrogenSegmentTestSample(base, 10, "source-a", true), nil)
+	for index := 1; index < 10; index++ {
+		mass := 10.0
+		if index >= 5 {
+			mass = 9.6
+		}
+		state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), mass, "source-a", true))
+	}
+	state.add(hydrogenSegmentTestSample(base.Add(10*time.Second), 12, "source-b", true))
+	for index := 11; index < 20; index++ {
+		mass := 12.0
+		if index >= 15 {
+			mass = 11.7
+		}
+		state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), mass, "source-b", true))
+	}
+	if got := state.projectedConsumptionKG(); math.Abs(got-0.7) > 0.0001 {
+		t.Fatalf("consumption=%.3f state=%#v", got, state)
+	}
+	if state.QualifiedSegmentCount != 1 || state.RefuelCount != 1 || state.Segment.Count != 10 {
+		t.Fatalf("state=%#v", state)
+	}
+}
+
+func TestHydrogenSegmentStreamFiltersAbnormalDrop(t *testing.T) {
+	base := time.Now()
+	state := newHydrogenSegmentStreamState(hydrogenSegmentTestSample(base, 30, "source-a", true), nil)
+	state.add(hydrogenSegmentTestSample(base.Add(time.Second), 5, "source-a", true))
+	for index := 2; index < 12; index++ {
+		state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), 5, "source-a", true))
+	}
+	quality, _ := state.quality()
+	if state.AbnormalDropCount != 1 || state.projectedConsumptionKG() != 0 || quality != "SUSPECT" {
+		t.Fatalf("state=%#v quality=%s", state, quality)
+	}
+}
+
+func TestHydrogenSegmentStreamSeedsDayEndBaselineWithoutRecounting(t *testing.T) {
+	base := time.Now()
+	baseline := &hydrogenStreamBaseline{
+		SourceEndpoint: "source-a", ConsumptionKG: 3.2, FirstMassKG: 14,
+		LastMassKG: 10, SampleCount: 100, RefuelCount: 1, QualityStatus: "OK",
+	}
+	state := newHydrogenSegmentStreamState(hydrogenSegmentTestSample(base, 10, "source-b", true), baseline)
+	for index := 1; index < 10; index++ {
+		mass := 10.0
+		if index >= 5 {
+			mass = 9.8
+		}
+		state.add(hydrogenSegmentTestSample(base.Add(time.Duration(index)*time.Second), mass, "source-b", true))
+	}
+	if got := state.projectedConsumptionKG(); math.Abs(got-3.4) > 0.0001 {
+		t.Fatalf("consumption=%.3f state=%#v", got, state)
+	}
+	if state.FirstMassKG != 14 || state.SampleCount != 110 || state.RefuelCount != 1 {
+		t.Fatalf("state=%#v", state)
+	}
+}
+
 func TestAppendHydrogenStreamPersistsPressureEvidenceAtomically(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	if err != nil {
@@ -72,14 +249,20 @@ func TestAppendHydrogenStreamPersistsPressureEvidenceAtomically(t *testing.T) {
 	loc := time.FixedZone("Asia/Shanghai", 8*3600)
 	eventTime := time.Date(2026, 7, 21, 10, 30, 0, 0, loc)
 	mock.ExpectBegin()
-	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_open_hydrogen_stream_state")).
-		WithArgs("LA9GG64L0NBAF4175", "2026-07-21", "10.0.0.1:9000", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(),
-			2050.0, 6.0, 6.0, 44.0, 44.0, eventTime, eventTime, "event-1",
-			sqlmock.AnyArg(), 20.0, sqlmock.AnyArg(), 20.0, 20.0, 20.0,
-			sqlmock.AnyArg(), sqlmock.AnyArg(), 20.0).
+	mock.ExpectQuery(regexp.QuoteMeta("SELECT state_json")).
+		WithArgs("LA9GG64L0NBAF4175", "2026-07-21").
+		WillReturnError(sql.ErrNoRows)
+	mock.ExpectQuery(regexp.QuoteMeta("SELECT source_endpoint,consumption_kg,first_mass_kg,last_mass_kg,")).
+		WithArgs("LA9GG64L0NBAF4175", "2026-07-21").
+		WillReturnError(sql.ErrNoRows)
+	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_open_hydrogen_segment_stream_state")).
+		WithArgs("LA9GG64L0NBAF4175", "2026-07-21", "10.0.0.1:9000",
+			sqlmock.AnyArg(), sqlmock.AnyArg(), int64(1), int64(0), int64(0), int64(0), int64(0),
+			sqlmock.AnyArg(), eventTime, "event-1", sqlmock.AnyArg(), "NO_DATA", "有效车载氢量样本不足2条").
 		WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_open_daily_energy")).
-		WithArgs("LA9GG64L0NBAF4175", "2026-07-21").
+		WithArgs("LA9GG64L0NBAF4175", "2026-07-21", "10.0.0.1:9000", 0.0,
+			sqlmock.AnyArg(), sqlmock.AnyArg(), int64(1), int64(0), "NO_DATA", "有效车载氢量样本不足2条").
 		WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit()

@@ -104,7 +287,9 @@ func TestWriterAppendWithResultUsesInMemoryCapacity(t *testing.T) {
 	loc := time.FixedZone("Asia/Shanghai", 8*3600)
 	eventTime := time.Now().In(loc).Truncate(time.Millisecond)
 	mock.ExpectBegin()
-	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_open_hydrogen_stream_state")).WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectQuery(regexp.QuoteMeta("SELECT state_json")).WillReturnError(sql.ErrNoRows)
+	mock.ExpectQuery(regexp.QuoteMeta("SELECT source_endpoint,consumption_kg,first_mass_kg,last_mass_kg,")).WillReturnError(sql.ErrNoRows)
+	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_open_hydrogen_segment_stream_state")).WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO vehicle_open_daily_energy")).WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit()

@@ -122,17 +307,17 @@ func TestWriterAppendWithResultUsesInMemoryCapacity(t *testing.T) {
 	}
 }

-func TestHydrogenStreamSQLGuardsDuplicatesAndRecordsPressure(t *testing.T) {
+func TestHydrogenStreamSQLUsesVINDateStateAndSegmentMethod(t *testing.T) {
 	for _, want := range []string{
-		"VALUES(last_event_time)>last_event_time",
-		"cycle_min_mass_kg-VALUES(last_mass_kg)>?",
-		"tank_capacity_l",
-		"last_pressure_mpa",
-		"last_temperature_c",
-		"PRESSURE_NIST",
+		"PRIMARY KEY (vin, stat_date)",
+		"state_json JSON NOT NULL",
+		"PRESSURE_NIST_SEGMENT_MEDIAN_5",
 	} {
-		if !regexp.MustCompile(regexp.QuoteMeta(want)).MatchString(upsertHydrogenStreamStateSQL) {
-			t.Fatalf("stream upsert missing %q", want)
+		if !strings.Contains(HydrogenSegmentStreamStateTableSQL, want) {
+			t.Fatalf("segment stream schema missing %q", want)
 		}
 	}
+	if !strings.Contains(selectHydrogenSegmentStreamStateSQL, "FOR UPDATE") {
+		t.Fatal("segment state must be locked before update")
+	}
 }
@@ -15,14 +15,40 @@ const (
 	yutongFuelCellDrivingMode     = 11
 )

-// PureHydrogenModeFromEnvelope returns whether the reported vehicle is in a
-// fuel-cell-only work mode and whether the source value is understood.
+// PureHydrogenModeFromEnvelope returns whether hydrogen power participates in
+// vehicle propulsion and whether the source value is understood.
 //
-// The 32960 fuel-cell extension reports engine_work_state; production frames
-// use mode 2 for normal fuel-cell operation. Yutong reports the equivalent
-// state through TRIANGLE_STATE; production frames use 1 for inactive and 4/11
-// for active fuel-cell operating modes.
+// GB32960's standard vehicle running mode is authoritative when present:
+// mode 1 is pure electric, mode 2 is hybrid, and mode 3 is fuel powered. Mode
+// 2 is counted as hydrogen-participating mileage for the hydrogen fleet; mode
+// 1 is excluded. Fuel, abnormal, and invalid values are left unclassified.
+// Older frames without the standard field fall back to the fuel-cell extension
+// engine_work_state. Yutong reports the equivalent state through
+// TRIANGLE_STATE; production frames use 1 for inactive and 4/11 for active.
 func PureHydrogenModeFromEnvelope(env envelope.FrameEnvelope) (active bool, known bool) {
+	if env.Protocol == envelope.ProtocolGB32960 {
+		if value, exists := env.Fields[envelope.FieldVehicleRunningMode]; exists {
+			mode, ok := integerMode(value)
+			if !ok {
+				return false, false
+			}
+			switch mode {
+			case 1:
+				return false, true
+			case 2:
+				return true, true
+			default:
+				return false, false
+			}
+		}
+	}
+	return FuelCellActiveModeFromEnvelope(env)
+}
+
+// FuelCellActiveModeFromEnvelope reads the fuel-cell subsystem work state.
+// Hydrogen consumption segmentation uses this signal (and its current-based
+// fallback) independently from the whole-vehicle running mode used by mileage.
+func FuelCellActiveModeFromEnvelope(env envelope.FrameEnvelope) (active bool, known bool) {
 	value, exists := env.Fields[envelope.FieldFuelCellWorkMode]
 	if !exists {
 		return false, false
@@ -56,8 +82,9 @@ func PureHydrogenModeFromEnvelope(env envelope.FrameEnvelope) (active bool, know
 }

 // PureHydrogenMileageDelta returns the odometer delta that can be attributed
-// to pure-hydrogen operation. Both endpoints must report a known active mode;
-// this deliberately leaves transitions and missing mode data unclassified.
+// to hydrogen-participating operation. The current endpoint owns the interval,
+// matching the daily rule: total mileage minus intervals ending in pure-electric
+// mode. Missing or invalid current modes remain unclassified.
 func PureHydrogenMileageDelta(
 	previousKM float64,
 	currentKM float64,
@@ -67,7 +94,7 @@ func PureHydrogenMileageDelta(
 	currentKnown bool,
 ) float64 {
 	delta := currentKM - previousKM
-	if !previousKnown || !previousActive || !currentKnown || !currentActive {
+	if !currentKnown || !currentActive {
 		return 0
 	}
 	if delta < 0 || delta > maxSelectedDailyMileageKM {
@@ -15,14 +15,24 @@ func TestPureHydrogenModeFromEnvelope(t *testing.T) {
 		known  bool
 	}{
 		{
-			name:   "gb32960 fuel cell mode",
-			env:    envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, Fields: map[string]any{envelope.FieldFuelCellWorkMode: 2}},
+			name:   "gb32960 hybrid running mode",
+			env:    envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, Fields: map[string]any{envelope.FieldVehicleRunningMode: 2}},
 			active: true,
 			known:  true,
 		},
 		{
 			name:  "gb32960 electric mode",
-			env:   envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, Fields: map[string]any{envelope.FieldFuelCellWorkMode: 1}},
+			env:   envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, Fields: map[string]any{envelope.FieldVehicleRunningMode: 1, envelope.FieldFuelCellWorkMode: 2}},
+			known: true,
+		},
+		{
+			name: "gb32960 fuel running mode is not hydrogen",
+			env:  envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, Fields: map[string]any{envelope.FieldVehicleRunningMode: 3}},
+		},
+		{
+			name:   "gb32960 extension fallback",
+			env:    envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, Fields: map[string]any{envelope.FieldFuelCellWorkMode: 2}},
+			active: true,
 			known:  true,
 		},
 		{
@@ -62,7 +72,7 @@ func TestPureHydrogenModeFromEnvelope(t *testing.T) {
 	}
 }

-func TestPureHydrogenMileageDeltaRequiresContinuousKnownActiveMode(t *testing.T) {
+func TestPureHydrogenMileageDeltaUsesCurrentKnownActiveMode(t *testing.T) {
 	tests := []struct {
 		name                          string
 		previousKM, currentKM         float64
@@ -81,15 +91,25 @@ func TestPureHydrogenMileageDeltaRequiresContinuousKnownActiveMode(t *testing.T)
 			want:           6.5,
 		},
 		{
-			name:          "transition into active is unclassified",
+			name:          "transition into active belongs to active interval",
 			previousKM:    100,
 			currentKM:     106.5,
 			previousKnown: true,
 			currentActive: true,
 			currentKnown:  true,
+			want:          6.5,
 		},
 		{
-			name:           "unknown endpoint is unclassified",
+			name:           "previous endpoint can be unknown",
+			previousKM:     100,
+			currentKM:      106.5,
+			previousActive: true,
+			currentActive:  true,
+			currentKnown:   true,
+			want:           6.5,
+		},
+		{
+			name:           "unknown current endpoint is unclassified",
 			previousKM:     100,
 			currentKM:      106.5,
 			previousActive: true,
@@ -8,6 +8,7 @@ const DailyMileageTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_mileage (
  daily_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0,
  pure_hydrogen_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0,
  latest_total_mileage_km DECIMAL(18,3) NULL,
+  day_end_total_mileage_km DECIMAL(18,3) NULL,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (vin, stat_date, protocol),
  KEY idx_stat_date (stat_date),
@@ -22,6 +23,7 @@ var DailyMileageAlterSQL = []string{
 	"ALTER TABLE vehicle_data_source ADD KEY idx_protocol_source_kind_seen (protocol, source_kind, latest_seen_at)",
 	"ALTER TABLE vehicle_daily_mileage ADD COLUMN source_id BIGINT NULL AFTER protocol",
 	"ALTER TABLE vehicle_daily_mileage ADD COLUMN pure_hydrogen_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0 AFTER daily_mileage_km",
+	"ALTER TABLE vehicle_daily_mileage ADD COLUMN day_end_total_mileage_km DECIMAL(18,3) NULL AFTER latest_total_mileage_km",
 	"ALTER TABLE vehicle_daily_mileage ADD KEY idx_source_id (source_id)",
 	"ALTER TABLE vehicle_daily_mileage DROP COLUMN first_total_mileage_km",
 	"ALTER TABLE vehicle_daily_mileage DROP COLUMN trusted_source_key",
@@ -36,6 +38,8 @@ var DailyMileageAlterSQL = []string{
 	"ALTER TABLE vehicle_daily_mileage_source ADD COLUMN pure_hydrogen_sample_count BIGINT NOT NULL DEFAULT 0 AFTER sample_count",
 	"ALTER TABLE vehicle_daily_mileage_source ADD COLUMN latest_pure_hydrogen_active TINYINT(1) NOT NULL DEFAULT 0 AFTER latest_event_time",
 	"ALTER TABLE vehicle_daily_mileage_source ADD COLUMN latest_pure_hydrogen_mode_known TINYINT(1) NOT NULL DEFAULT 0 AFTER latest_pure_hydrogen_active",
+	"ALTER TABLE vehicle_open_hydrogen_stream_state ADD COLUMN last_fuel_cell_active TINYINT(1) NOT NULL DEFAULT 0 AFTER last_event_id",
+	"ALTER TABLE vehicle_open_hydrogen_stream_state ADD COLUMN last_fuel_cell_state_known TINYINT(1) NOT NULL DEFAULT 0 AFTER last_fuel_cell_active",
 }

 const DataSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_data_source (
@@ -13,6 +13,7 @@ const (
 	QualityOK                     = "OK"
 	QualityNoPreviousBaseline     = "NO_PREVIOUS_BASELINE"
 	QualityInvalidDelta           = "INVALID_DELTA"
+	QualityReasonTotalRollback    = "TOTAL_MILEAGE_ROLLBACK"
 	QualityReasonHistorical       = "historical_source_baseline"
 	QualityReasonCurrentDayFirst  = "current_day_first_baseline"
 	QualityReasonStationaryCarry  = "stationary_carry_forward"
@@ -136,6 +137,26 @@ func DailyMileageFromDayBoundary(previousBaselineKM float64, currentDayLatestKM
 	return currentDayLatestKM - previousBaselineKM
 }

+// IsUsableDailyMileageBoundary reports whether a persisted baseline belongs to
+// the current natural day or its immediately preceding natural day. A baseline
+// from an older day represents an offline gap; using it would attribute the
+// whole gap to the day on which the vehicle came back online.
+func IsUsableDailyMileageBoundary(statDate string, baselineTime time.Time, loc *time.Location) bool {
+	if strings.TrimSpace(statDate) == "" || baselineTime.IsZero() {
+		return false
+	}
+	if loc == nil {
+		loc = baselineTime.Location()
+	}
+	day, err := time.ParseInLocation("2006-01-02", statDate, loc)
+	if err != nil {
+		return false
+	}
+	baseline := baselineTime.In(loc)
+	baselineDay := time.Date(baseline.Year(), baseline.Month(), baseline.Day(), 0, 0, 0, 0, loc)
+	return !baselineDay.Before(day.AddDate(0, 0, -1)) && !baselineDay.After(day)
+}
+
 func NormalizeDailyMileageDeltaForWindow(deltaKM float64, firstEventTime time.Time, latestEventTime time.Time) (float64, bool, string) {
 	if deltaKM < 0 && deltaKM >= -maxNegativeMileageJitterKM {
 		return 0, true, "negative_jitter_clamped"
@@ -444,8 +465,6 @@ const normalizePlatformOutsideDailyRangeSQL = `(` + normalizePlatformDailySQL +
 const upsertSourcePureHydrogenIncrementSQL = `CASE
    WHEN latest_event_time IS NOT NULL
     AND VALUES(latest_event_time) > latest_event_time
-     AND latest_pure_hydrogen_mode_known = 1
-     AND latest_pure_hydrogen_active = 1
     AND VALUES(latest_pure_hydrogen_mode_known) = 1
     AND VALUES(latest_pure_hydrogen_active) = 1
     AND VALUES(latest_total_mileage_km) >= latest_total_mileage_km
@@ -731,9 +750,32 @@ const selectedSourceIdentitySampleCountSQL = `(
      = COALESCE(NULLIF(TRIM(s2.phone), ''), NULLIF(TRIM(s2.device_id), ''), s2.source_key)
 )`

+// projectDayEndTotalMileageSQL intentionally uses an independent source choice
+// from the daily-mileage projection. GPS coordinate accumulation can be the
+// best evidence for distance travelled during the day, but it must never
+// replace a terminal-reported odometer as the day-end cumulative mileage.
+const projectDayEndTotalMileageSQL = `COALESCE((
+  SELECT total_candidate.latest_total_mileage_km
+  FROM vehicle_daily_mileage_source total_candidate
+  WHERE total_candidate.vin = s.vin
+    AND total_candidate.stat_date = s.stat_date
+    AND total_candidate.protocol = s.protocol
+    AND total_candidate.quality_status = '` + QualityOK + `'
+    AND total_candidate.latest_total_mileage_km IS NOT NULL
+    AND total_candidate.latest_total_mileage_km >= 0
+  ORDER BY
+    CASE WHEN COALESCE(total_candidate.quality_reason, '') = '` + QualityReasonGPSCoordinate + `' THEN 1 ELSE 0 END,
+    CASE WHEN total_candidate.latest_total_mileage_km > total_candidate.daily_mileage_km THEN 0 ELSE 1 END,
+    total_candidate.is_selected DESC,
+    total_candidate.latest_event_time DESC,
+    total_candidate.sample_count DESC,
+    total_candidate.source_key ASC
+  LIMIT 1
+), s.latest_total_mileage_km)`
+
 const projectDailyMileageSQL = `
 INSERT INTO vehicle_daily_mileage
-  (vin, stat_date, protocol, source_id, daily_mileage_km, pure_hydrogen_mileage_km, latest_total_mileage_km)
+  (vin, stat_date, protocol, source_id, daily_mileage_km, pure_hydrogen_mileage_km, latest_total_mileage_km, day_end_total_mileage_km)
 SELECT
  s.vin,
  s.stat_date,
@@ -741,7 +783,8 @@ SELECT
  ds.id,
  s.daily_mileage_km,
  LEAST(s.pure_hydrogen_mileage_km, s.daily_mileage_km),
-  s.latest_total_mileage_km
+  s.latest_total_mileage_km,
+  ` + projectDayEndTotalMileageSQL + `
 FROM vehicle_daily_mileage_source s
 LEFT JOIN vehicle_data_source ds
  ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
@@ -774,6 +817,7 @@ ON DUPLICATE KEY UPDATE
  daily_mileage_km = VALUES(daily_mileage_km),
  pure_hydrogen_mileage_km = VALUES(pure_hydrogen_mileage_km),
  latest_total_mileage_km = VALUES(latest_total_mileage_km),
+  day_end_total_mileage_km = VALUES(day_end_total_mileage_km),
  updated_at = CURRENT_TIMESTAMP
 `

@@ -866,6 +910,35 @@ func lookupPreviousSourceBaseline(ctx context.Context, query Queryer, vin string
 	}, latestTotal.Valid, nil
 }

+// lookupPreviousProtocolBaseline is deliberately source-key independent. A
+// device replacement must not allow a near-zero odometer to become a normal
+// continuation of the same vehicle/protocol history.
+func lookupPreviousProtocolBaseline(ctx context.Context, query Queryer, vin string, statDate string, protocol envelope.Protocol) (sourceBaseline, bool, error) {
+	if query == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" {
+		return sourceBaseline{}, false, nil
+	}
+	rows, err := query.QueryContext(ctx, previousProtocolBaselineSQL, vin, statDate, string(protocol))
+	if err != nil {
+		return sourceBaseline{}, false, err
+	}
+	defer rows.Close()
+	if !rows.Next() {
+		if err := rows.Err(); err != nil {
+			return sourceBaseline{}, false, err
+		}
+		return sourceBaseline{}, false, nil
+	}
+	var total sql.NullFloat64
+	var event sql.NullTime
+	if err := rows.Scan(&total, &event); err != nil {
+		return sourceBaseline{}, false, err
+	}
+	if err := rows.Err(); err != nil {
+		return sourceBaseline{}, false, err
+	}
+	return sourceBaseline{LatestTotalKM: total.Float64, LatestEventTime: event.Time, QualityReason: QualityReasonHistorical}, total.Valid, nil
+}
+
 // LookupLatestSourceBaselineBefore returns the nearest durable odometer for the
 // same VIN, protocol and source before statDate. Missing calendar days are
 // skipped automatically.
@@ -889,3 +962,14 @@ WHERE vin = ?
 ORDER BY stat_date DESC, latest_event_time DESC
 LIMIT 1
 `
+
+const previousProtocolBaselineSQL = `
+SELECT latest_total_mileage_km, latest_event_time
+FROM vehicle_daily_mileage_source
+WHERE vin=? AND stat_date<? AND protocol=?
+  AND quality_status='OK'
+  AND latest_total_mileage_km IS NOT NULL AND latest_total_mileage_km>0
+  AND latest_event_time IS NOT NULL
+ORDER BY stat_date DESC, latest_event_time DESC
+LIMIT 1
+`
@@ -211,12 +211,10 @@ func TestUpsertSourceMileageUsesEventTimeBoundaries(t *testing.T) {
 	}
 }

-func TestUpsertSourceMileageAccumulatesOnlyContinuousPureHydrogenIntervals(t *testing.T) {
+func TestUpsertSourceMileageAttributesDeltaToCurrentPureHydrogenMode(t *testing.T) {
 	for _, want := range []string{
 		"pure_hydrogen_mileage_km = pure_hydrogen_mileage_km + CASE",
 		"VALUES(latest_event_time) > latest_event_time",
-		"latest_pure_hydrogen_mode_known = 1",
-		"latest_pure_hydrogen_active = 1",
 		"VALUES(latest_pure_hydrogen_mode_known) = 1",
 		"VALUES(latest_pure_hydrogen_active) = 1",
 		"VALUES(latest_total_mileage_km) - latest_total_mileage_km",
@@ -281,6 +279,26 @@ func TestDailyMileageFromDayBoundaryUsesCurrentMinusPrevious(t *testing.T) {
 	}
 }

+func TestIsUsableDailyMileageBoundaryRejectsOlderOfflineGap(t *testing.T) {
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	for _, test := range []struct {
+		name     string
+		baseline time.Time
+		want     bool
+	}{
+		{name: "current day cache", baseline: time.Date(2026, 8, 4, 19, 1, 26, 0, loc), want: true},
+		{name: "previous natural day", baseline: time.Date(2026, 8, 3, 23, 59, 59, 0, loc), want: true},
+		{name: "older offline gap", baseline: time.Date(2026, 5, 21, 23, 13, 2, 0, loc), want: false},
+		{name: "future boundary", baseline: time.Date(2026, 8, 5, 0, 0, 0, 0, loc), want: false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if got := IsUsableDailyMileageBoundary("2026-08-04", test.baseline, loc); got != test.want {
+				t.Fatalf("IsUsableDailyMileageBoundary() = %v, want %v", got, test.want)
+			}
+		})
+	}
+}
+
 func TestUpsertSourceMileageSkipsBlankSourceIP(t *testing.T) {
 	exec := &recordingExec{}
 	sample := SourceMileageSample{
@@ -612,6 +630,15 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
 		"project": projectFinal,
 		"mark":    markSelected,
 	} {
+		// The project statement also contains an independent day-end odometer
+		// subquery. Check the daily-mileage election order after its main source
+		// FROM clause, rather than mistaking the odometer fallback's freshness
+		// order for the elected daily-distance source order.
+		if queryName == "project" {
+			if index := strings.Index(query, "FROM vehicle_daily_mileage_source s\nLEFT JOIN"); index >= 0 {
+				query = query[index:]
+			}
+		}
 		identityScore := strings.Index(query, "SUM(identity_source.sample_count)")
 		freshness := strings.Index(query, "latest_event_time DESC")
 		rowSamples := strings.LastIndex(query, "sample_count DESC")
@@ -699,3 +726,16 @@ func expectProjectDailyMileageSQL(mock sqlmock.Sqlmock, vin string, statDate str
 		WithArgs(vin, statDate, string(protocol), vin, statDate, string(protocol)).
 		WillReturnResult(sqlmock.NewResult(0, 1))
 }
+
+func TestProjectDailyMileagePersistsIndependentDayEndTotal(t *testing.T) {
+	for _, want := range []string{
+		"day_end_total_mileage_km",
+		"total_candidate.quality_status = 'OK'",
+		"total_candidate.quality_reason, '') = 'gps_coordinate_accumulation'",
+		"day_end_total_mileage_km = VALUES(day_end_total_mileage_km)",
+	} {
+		if !strings.Contains(projectDailyMileageSQL, want) {
+			t.Fatalf("project SQL missing %q:\n%s", want, projectDailyMileageSQL)
+		}
+	}
+}
@@ -245,6 +245,16 @@ func TestDailyMileageSchemaIncludesPureHydrogenEvidence(t *testing.T) {
 	}
 }

+func TestDailyMileageSchemaIncludesDayEndTotalMileage(t *testing.T) {
+	const column = "day_end_total_mileage_km DECIMAL(18,3) NULL"
+	if !strings.Contains(DailyMileageTableSQL, column) {
+		t.Fatalf("daily mileage schema missing %q:\n%s", column, DailyMileageTableSQL)
+	}
+	if !containsStatement(DailyMileageAlterSQL, "ALTER TABLE vehicle_daily_mileage ADD COLUMN "+column) {
+		t.Fatalf("daily mileage alter SQL missing %q: %#v", column, DailyMileageAlterSQL)
+	}
+}
+
 func containsStatement(statements []string, fragment string) bool {
 	for _, statement := range statements {
 		if strings.Contains(statement, fragment) {
@@ -0,0 +1,126 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+BINARY="${BINARY:-/opt/lingniu-go-native/current/stats-backfill}"
+DATE_FROM="${DATE_FROM:-}"
+DATE_TO="${DATE_TO:-}"
+PARALLEL="${PARALLEL:-4}"
+APPLY=false
+WORK_DIR=""
+
+usage() {
+  cat <<'USAGE'
+Usage: replay-pure-hydrogen-mileage.sh --date-from YYYY-MM-DD --date-to YYYY-MM-DD [options]
+
+Options:
+  --apply             Write the rebuilt daily mileage rows. Default is dry-run.
+  --parallel N        Concurrent VIN workers. Default: 4.
+  --binary PATH       stats-backfill binary path.
+  --work-dir PATH     Keep VIN lists and per-VIN logs in this directory.
+USAGE
+}
+
+if [[ "${1:-}" == "--worker" ]]; then
+  vin="$2"
+  dry_run="$3"
+  log_dir="$4"
+  env \
+    BACKFILL_METHOD=scan \
+    BACKFILL_DRY_RUN="$dry_run" \
+    BACKFILL_DATE_FROM="$DATE_FROM" \
+    BACKFILL_DATE_TO="$DATE_TO" \
+    BACKFILL_PROTOCOLS=GB32960 \
+    BACKFILL_VINS="$vin" \
+    BACKFILL_PROGRESS_EVERY=100000 \
+    "$BINARY" >"$log_dir/$vin.log" 2>&1
+  printf 'completed vin=%s\n' "$vin"
+  exit 0
+fi
+
+while (($#)); do
+  case "$1" in
+    --date-from) DATE_FROM="$2"; shift 2 ;;
+    --date-to) DATE_TO="$2"; shift 2 ;;
+    --parallel) PARALLEL="$2"; shift 2 ;;
+    --binary) BINARY="$2"; shift 2 ;;
+    --work-dir) WORK_DIR="$2"; shift 2 ;;
+    --apply) APPLY=true; shift ;;
+    -h|--help) usage; exit 0 ;;
+    *) echo "unknown argument: $1" >&2; usage >&2; exit 2 ;;
+  esac
+done
+
+if [[ ! "$DATE_FROM" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ || ! "$DATE_TO" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]]; then
+  echo "--date-from and --date-to are required" >&2
+  exit 2
+fi
+if [[ ! "$PARALLEL" =~ ^[1-9][0-9]*$ ]]; then
+  echo "--parallel must be a positive integer" >&2
+  exit 2
+fi
+if [[ ! -x "$BINARY" ]]; then
+  echo "stats-backfill is not executable: $BINARY" >&2
+  exit 2
+fi
+
+if [[ -z "$WORK_DIR" ]]; then
+  WORK_DIR="$(mktemp -d /tmp/pure-hydrogen-replay.XXXXXX)"
+else
+  mkdir -p "$WORK_DIR"
+fi
+LOG_DIR="$WORK_DIR/logs"
+mkdir -p "$LOG_DIR"
+
+hydrogen_vins="$WORK_DIR/hydrogen-vins.txt"
+active_vins="$WORK_DIR/active-gb32960-vins.txt"
+target_vins="$WORK_DIR/target-vins.txt"
+pending_vins="$WORK_DIR/pending-vins.txt"
+
+BACKFILL_LIST_HYDROGEN_VINS=true "$BINARY" | LC_ALL=C sort -u >"$hydrogen_vins"
+env \
+  BACKFILL_LIST_ACTIVE_VINS=true \
+  BACKFILL_DATE_FROM="$DATE_FROM" \
+  BACKFILL_DATE_TO="$DATE_TO" \
+  BACKFILL_PROTOCOLS=GB32960 \
+  "$BINARY" | LC_ALL=C sort -u >"$active_vins"
+comm -12 "$hydrogen_vins" "$active_vins" >"$target_vins"
+
+target_count="$(wc -l <"$target_vins" | tr -d ' ')"
+: >"$pending_vins"
+while IFS= read -r vin; do
+  if [[ -f "$LOG_DIR/$vin.log" ]] && grep -q 'stats backfill complete' "$LOG_DIR/$vin.log"; then
+    continue
+  fi
+  echo "$vin" >>"$pending_vins"
+done <"$target_vins"
+pending_count="$(wc -l <"$pending_vins" | tr -d ' ')"
+dry_run=true
+if [[ "$APPLY" == true ]]; then
+  dry_run=false
+fi
+printf 'pure-hydrogen replay: date=%s..%s vins=%s pending=%s parallel=%s dry_run=%s work_dir=%s\n' \
+  "$DATE_FROM" "$DATE_TO" "$target_count" "$pending_count" "$PARALLEL" "$dry_run" "$WORK_DIR"
+
+if [[ "$pending_count" == "0" ]]; then
+  printf 'pure-hydrogen replay complete: vins=%s logs=%s\n' "$target_count" "$LOG_DIR"
+  exit 0
+fi
+
+export BINARY DATE_FROM DATE_TO
+set +e
+xargs -I '{}' -P "$PARALLEL" "$0" --worker '{}' "$dry_run" "$LOG_DIR" <"$pending_vins"
+xargs_status=$?
+set -e
+
+failed_vins="$WORK_DIR/failed-vins.txt"
+: >"$failed_vins"
+while IFS= read -r vin; do
+  if [[ ! -f "$LOG_DIR/$vin.log" ]] || ! grep -q 'stats backfill complete' "$LOG_DIR/$vin.log"; then
+    echo "$vin" >>"$failed_vins"
+  fi
+done <"$target_vins"
+if [[ "$xargs_status" != "0" || -s "$failed_vins" ]]; then
+  echo "replay incomplete; inspect $failed_vins and $LOG_DIR" >&2
+  exit 1
+fi
+printf 'pure-hydrogen replay complete: vins=%s logs=%s\n' "$target_count" "$LOG_DIR"
@@ -0,0 +1,29 @@
+-- Persist the end-of-day cumulative odometer independently of the source that
+-- wins the daily-distance projection. This prevents GPS-derived daily mileage
+-- from being exposed as terminal odometer mileage.
+ALTER TABLE vehicle_daily_mileage
+    ADD COLUMN day_end_total_mileage_km DECIMAL(18,3) NULL AFTER latest_total_mileage_km;
+
+-- Backfill every historical day with the best valid terminal/source cumulative
+-- reading. Non-GPS evidence always wins; GPS remains the fallback only where
+-- it is the sole available cumulative source.
+UPDATE vehicle_daily_mileage m
+SET m.day_end_total_mileage_km = COALESCE((
+    SELECT total_candidate.latest_total_mileage_km
+    FROM vehicle_daily_mileage_source total_candidate
+    WHERE total_candidate.vin=m.vin
+      AND total_candidate.stat_date=m.stat_date
+      AND total_candidate.protocol=m.protocol
+      AND total_candidate.quality_status='OK'
+      AND total_candidate.latest_total_mileage_km IS NOT NULL
+      AND total_candidate.latest_total_mileage_km>=0
+    ORDER BY
+      CASE WHEN COALESCE(total_candidate.quality_reason,'')='gps_coordinate_accumulation' THEN 1 ELSE 0 END,
+      CASE WHEN total_candidate.latest_total_mileage_km>total_candidate.daily_mileage_km THEN 0 ELSE 1 END,
+      total_candidate.is_selected DESC,
+      total_candidate.latest_event_time DESC,
+      total_candidate.sample_count DESC,
+      total_candidate.source_key ASC
+    LIMIT 1
+),m.latest_total_mileage_km)
+WHERE m.day_end_total_mileage_km IS NULL;
@@ -0,0 +1,24 @@
+CREATE TABLE IF NOT EXISTS vehicle_open_hydrogen_segment_stream_state (
+    vin VARCHAR(64) NOT NULL,
+    stat_date DATE NOT NULL,
+    source_endpoint VARCHAR(128) NOT NULL DEFAULT '',
+    finalized_consumption_kg DECIMAL(18,3) NOT NULL DEFAULT 0,
+    projected_consumption_kg DECIMAL(18,3) NOT NULL DEFAULT 0,
+    sample_count BIGINT UNSIGNED NOT NULL DEFAULT 0,
+    refuel_count INT UNSIGNED NOT NULL DEFAULT 0,
+    abnormal_drop_count INT UNSIGNED NOT NULL DEFAULT 0,
+    eligible_interval_count BIGINT UNSIGNED NOT NULL DEFAULT 0,
+    qualified_segment_count INT UNSIGNED NOT NULL DEFAULT 0,
+    last_mass_kg DECIMAL(18,3) NOT NULL DEFAULT 0,
+    last_event_time DATETIME(3) NOT NULL,
+    last_event_id VARCHAR(128) NOT NULL DEFAULT '',
+    state_json JSON NOT NULL,
+    calculation_method VARCHAR(48) NOT NULL DEFAULT 'PRESSURE_NIST_SEGMENT_MEDIAN_5',
+    quality_status VARCHAR(24) NOT NULL DEFAULT 'NO_DATA',
+    quality_reason VARCHAR(255) NOT NULL DEFAULT '有效工作段关键帧不足10条',
+    created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+    updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+    PRIMARY KEY (vin, stat_date),
+    INDEX idx_hydrogen_segment_stream_date_quality (stat_date, quality_status, vin),
+    CONSTRAINT chk_hydrogen_segment_stream_quality CHECK (quality_status IN ('OK', 'NO_DATA', 'SUSPECT'))
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
@@ -0,0 +1,34 @@
+CREATE TABLE IF NOT EXISTS vehicle_hydrogen_energy_parameter (
+    vin VARCHAR(64) NOT NULL,
+    battery_capacity_kwh DECIMAL(12,3) NOT NULL,
+    hydrogen_energy_kwh_per_kg DECIMAL(12,3) NOT NULL DEFAULT 16.000,
+    confirmed_by VARCHAR(96) NOT NULL DEFAULT '',
+    source_note VARCHAR(255) NOT NULL DEFAULT '',
+    effective_from DATE NOT NULL,
+    effective_to DATE NULL,
+    active TINYINT(1) NOT NULL DEFAULT 1,
+    created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+    updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+    PRIMARY KEY (vin, effective_from),
+    INDEX idx_hydrogen_energy_parameter_active (active, effective_from, effective_to, vin),
+    CONSTRAINT chk_hydrogen_energy_parameter_capacity CHECK (battery_capacity_kwh > 0 AND battery_capacity_kwh <= 500),
+    CONSTRAINT chk_hydrogen_energy_parameter_conversion CHECK (hydrogen_energy_kwh_per_kg > 0 AND hydrogen_energy_kwh_per_kg <= 50),
+    CONSTRAINT chk_hydrogen_energy_parameter_dates CHECK (effective_to IS NULL OR effective_to >= effective_from)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+ALTER TABLE vehicle_open_daily_energy
+    ADD COLUMN raw_consumption_kg DECIMAL(18,3) NULL AFTER consumption_kg,
+    ADD COLUMN battery_soc_delta_pct DECIMAL(10,3) NULL AFTER raw_consumption_kg,
+    ADD COLUMN battery_discharge_kwh DECIMAL(18,3) NULL AFTER battery_soc_delta_pct,
+    ADD COLUMN battery_equivalent_kg DECIMAL(18,3) NULL AFTER battery_discharge_kwh,
+    ADD COLUMN soc_balanced_consumption_kg DECIMAL(18,3) NULL AFTER battery_equivalent_kg,
+    ADD COLUMN mixed_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0 AFTER soc_balanced_consumption_kg,
+    ADD COLUMN pure_electric_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0 AFTER mixed_mileage_km,
+    ADD COLUMN consumption_kg_per_100km DECIMAL(18,3) NULL AFTER pure_electric_mileage_km,
+    ADD COLUMN soc_balanced_kg_per_100km DECIMAL(18,3) NULL AFTER consumption_kg_per_100km,
+    ADD COLUMN charge_count INT UNSIGNED NOT NULL DEFAULT 0 AFTER refuel_count,
+    ADD COLUMN valid_segment_count INT UNSIGNED NOT NULL DEFAULT 0 AFTER charge_count,
+    ADD COLUMN invalid_segment_count INT UNSIGNED NOT NULL DEFAULT 0 AFTER valid_segment_count,
+    ADD COLUMN algorithm_version VARCHAR(64) NOT NULL DEFAULT 'PRESSURE_NIST_SEGMENT_MEDIAN_5' AFTER invalid_segment_count,
+    ADD COLUMN parameter_json JSON NULL AFTER algorithm_version,
+    ADD COLUMN evidence_json JSON NULL AFTER parameter_json;