diff --git a/src/Core/Normalization/Normalizers/ConsumableBalanceNormalizer.cs b/src/Core/Normalization/Normalizers/ConsumableBalanceNormalizer.cs
index 0c252f0..c626286 100644
--- a/src/Core/Normalization/Normalizers/ConsumableBalanceNormalizer.cs
+++ b/src/Core/Normalization/Normalizers/ConsumableBalanceNormalizer.cs
@@ -21,7 +21,10 @@ public sealed class ConsumableBalanceNormalizer : IMeterNormalizer
var calibration = context.Meter.Tank?.Calibration;
var events = context.Events
.Where(e => e.EventType is MeterEventType.Delivery or MeterEventType.TankLevel)
+ // At the same instant a delivery is applied before the level is read, otherwise a
+ // refill reads as a negative draw and gets clamped away.
.OrderBy(e => e.Time)
+ .ThenBy(e => e.EventType == MeterEventType.Delivery ? 0 : 1)
.ToList();
double? lastLevelVolume = null;
diff --git a/src/Core/Parsing/GermanDate.cs b/src/Core/Parsing/GermanDate.cs
new file mode 100644
index 0000000..85eba72
--- /dev/null
+++ b/src/Core/Parsing/GermanDate.cs
@@ -0,0 +1,68 @@
+using System.Globalization;
+
+namespace MeterVault.Core.Parsing;
+
+///
+/// Parses the two date shapes in the reference data (SDD Appendix A): German month tables
+/// (September 2022) and event rows (10.06.1997, DD.MM.YYYY). Month tables are
+/// anchored to the first of the month.
+///
+public static class GermanDate
+{
+ private static readonly Dictionary MonthNames = new(StringComparer.OrdinalIgnoreCase)
+ {
+ ["Januar"] = 1,
+ ["Februar"] = 2,
+ ["März"] = 3,
+ ["Maerz"] = 3,
+ ["April"] = 4,
+ ["Mai"] = 5,
+ ["Juni"] = 6,
+ ["Juli"] = 7,
+ ["August"] = 8,
+ ["September"] = 9,
+ ["Oktober"] = 10,
+ ["November"] = 11,
+ ["Dezember"] = 12,
+ };
+
+ public static bool TryParseMonth(string? raw, out DateOnly firstOfMonth)
+ {
+ firstOfMonth = default;
+ if (string.IsNullOrWhiteSpace(raw))
+ {
+ return false;
+ }
+
+ var parts = raw.Trim().Split(' ', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
+ if (parts.Length != 2
+ || !MonthNames.TryGetValue(parts[0], out var month)
+ || !int.TryParse(parts[1], NumberStyles.Integer, CultureInfo.InvariantCulture, out var year))
+ {
+ return false;
+ }
+
+ firstOfMonth = new DateOnly(year, month, 1);
+ return true;
+ }
+
+ public static bool TryParseDay(string? raw, out DateOnly date) =>
+ DateOnly.TryParseExact(raw?.Trim(), "dd.MM.yyyy", CultureInfo.InvariantCulture, DateTimeStyles.None, out date);
+
+ /// Tries day format first, then month-name format (SDD: oil switches mid-file).
+ public static bool TryParse(string? raw, out DateOnly date)
+ {
+ if (TryParseDay(raw, out date))
+ {
+ return true;
+ }
+
+ if (TryParseMonth(raw, out date))
+ {
+ return true;
+ }
+
+ date = default;
+ return false;
+ }
+}
diff --git a/src/Core/Parsing/GermanMoney.cs b/src/Core/Parsing/GermanMoney.cs
new file mode 100644
index 0000000..17e9ab5
--- /dev/null
+++ b/src/Core/Parsing/GermanMoney.cs
@@ -0,0 +1,62 @@
+using System.Globalization;
+
+namespace MeterVault.Core.Parsing;
+
+///
+/// Parses German currency cells (SDD Appendix A): trailing € with a space, decimal comma,
+/// thousands dot (120,00 € → (120.00, "EUR"), 2.940,19 € → (2940.19, "EUR"),
+/// 70 € → (70, "EUR")). Amount is a decimal; currency defaults to EUR.
+///
+public static class GermanMoney
+{
+ public static bool TryParse(ReadOnlySpan raw, out decimal amount, out string currency)
+ {
+ amount = 0m;
+ currency = "EUR";
+ raw = raw.Trim();
+ if (raw.IsEmpty)
+ {
+ return false;
+ }
+
+ var sign = 1;
+ var index = 0;
+ if (raw[0] is '-' or '+')
+ {
+ sign = raw[0] == '-' ? -1 : 1;
+ index = 1;
+ }
+
+ var start = index;
+ while (index < raw.Length && (char.IsDigit(raw[index]) || raw[index] is '.' or ','))
+ {
+ index++;
+ }
+
+ var numeric = raw[start..index];
+ if (numeric.IsEmpty)
+ {
+ return false;
+ }
+
+ Span buffer = stackalloc char[numeric.Length];
+ var length = 0;
+ foreach (var c in numeric)
+ {
+ if (c == '.')
+ {
+ continue;
+ }
+
+ buffer[length++] = c == ',' ? '.' : c;
+ }
+
+ if (!decimal.TryParse(buffer[..length], NumberStyles.Float, CultureInfo.InvariantCulture, out var parsed))
+ {
+ return false;
+ }
+
+ amount = sign * parsed;
+ return true;
+ }
+}
diff --git a/src/Core/Parsing/GermanNumber.cs b/src/Core/Parsing/GermanNumber.cs
new file mode 100644
index 0000000..59f8ff4
--- /dev/null
+++ b/src/Core/Parsing/GermanNumber.cs
@@ -0,0 +1,66 @@
+using System.Globalization;
+
+namespace MeterVault.Core.Parsing;
+
+///
+/// Parses the reference spreadsheet's German numeric dialect (SDD Appendix A): decimal comma,
+/// thousands dot, optional sign, and an optional trailing unit/currency suffix which is ignored
+/// (411kWh → 411, 2.940,19 € → 2940.19, -90kWh → -90, 180,8244706 →
+/// 180.82…). Done by explicit transform rather than a de-DE NumberStyles parse because of the
+/// mixed unit suffixes.
+///
+public static class GermanNumber
+{
+ public static bool TryParse(ReadOnlySpan raw, out double value)
+ {
+ value = 0;
+ raw = raw.Trim();
+ if (raw.IsEmpty)
+ {
+ return false;
+ }
+
+ var sign = 1;
+ var index = 0;
+ if (raw[0] is '-' or '+')
+ {
+ sign = raw[0] == '-' ? -1 : 1;
+ index = 1;
+ }
+
+ var start = index;
+ while (index < raw.Length && (char.IsDigit(raw[index]) || raw[index] is '.' or ','))
+ {
+ index++;
+ }
+
+ var numeric = raw[start..index];
+ if (numeric.IsEmpty)
+ {
+ return false;
+ }
+
+ // Strip thousands dots, turn the decimal comma into a point, then parse invariant.
+ Span buffer = stackalloc char[numeric.Length];
+ var length = 0;
+ foreach (var c in numeric)
+ {
+ if (c == '.')
+ {
+ continue;
+ }
+
+ buffer[length++] = c == ',' ? '.' : c;
+ }
+
+ if (!double.TryParse(buffer[..length], NumberStyles.Float, CultureInfo.InvariantCulture, out var parsed))
+ {
+ return false;
+ }
+
+ value = sign * parsed;
+ return true;
+ }
+
+ public static double? Parse(ReadOnlySpan raw) => TryParse(raw, out var value) ? value : null;
+}
diff --git a/src/Core/Parsing/ValueCell.cs b/src/Core/Parsing/ValueCell.cs
new file mode 100644
index 0000000..0420b90
--- /dev/null
+++ b/src/Core/Parsing/ValueCell.cs
@@ -0,0 +1,33 @@
+namespace MeterVault.Core.Parsing;
+
+///
+/// Splits a value cell that carries a trailing unit into its number and unit
+/// (SDD Appendix A): 2287L → (2287, "L"), 411kWh → (411, "kWh"), 49 →
+/// (49, null). The unit is validated by the caller against the meter's expected unit.
+///
+public static class ValueCell
+{
+ public static (double Value, string? Unit)? Split(ReadOnlySpan raw)
+ {
+ raw = raw.Trim();
+ if (raw.IsEmpty)
+ {
+ return null;
+ }
+
+ var sign = raw[0] is '-' or '+' ? 1 : 0;
+ var index = sign;
+ while (index < raw.Length && (char.IsDigit(raw[index]) || raw[index] is '.' or ','))
+ {
+ index++;
+ }
+
+ if (!GermanNumber.TryParse(raw[..index], out var value))
+ {
+ return null;
+ }
+
+ var unit = raw[index..].Trim();
+ return (value, unit.IsEmpty ? null : unit.ToString());
+ }
+}
diff --git a/src/Infrastructure/DependencyInjection.cs b/src/Infrastructure/DependencyInjection.cs
index 214e84c..d196a80 100644
--- a/src/Infrastructure/DependencyInjection.cs
+++ b/src/Infrastructure/DependencyInjection.cs
@@ -1,3 +1,6 @@
+using MeterVault.Core.Normalization;
+using MeterVault.Infrastructure.Import;
+using MeterVault.Infrastructure.Normalization;
using MeterVault.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
@@ -17,6 +20,11 @@ public static class DependencyInjection
npgsql.MigrationsAssembly(typeof(MeterVaultDbContext).Assembly.FullName))
.UseSnakeCaseNamingConvention());
+ services.AddSingleton(_ => NormalizationEngine.CreateDefault());
+ services.AddScoped();
+ services.AddScoped();
+ services.AddScoped();
+
return services;
}
}
diff --git a/src/Infrastructure/Import/CsvImporter.cs b/src/Infrastructure/Import/CsvImporter.cs
new file mode 100644
index 0000000..df45991
--- /dev/null
+++ b/src/Infrastructure/Import/CsvImporter.cs
@@ -0,0 +1,244 @@
+using System.Globalization;
+using CsvHelper;
+using CsvHelper.Configuration;
+using MeterVault.Core.Domain;
+using MeterVault.Core.Parsing;
+
+namespace MeterVault.Infrastructure.Import;
+
+///
+/// Reads a CSV in the reference German dialect and stages readings/events/manual-costs per a
+/// (SDD §6.3). Pure transformation — no persistence; feed the result
+/// to to commit as a revertible batch, or to the normalization engine
+/// for a dry-run preview.
+///
+public sealed class CsvImporter
+{
+ public StagedImport Stage(MappingProfile profile, TextReader reader)
+ {
+ ArgumentNullException.ThrowIfNull(profile);
+ ArgumentNullException.ThrowIfNull(reader);
+
+ var rows = ReadRows(reader);
+ var staged = new StagedImport();
+ var previousByMeter = new Dictionary();
+
+ for (var r = profile.FirstDataRowIndex; r < rows.Count; r++)
+ {
+ var row = rows[r];
+ if (RowClassifier.ShouldSkip(row, profile, out var reason))
+ {
+ staged.SkippedRows++;
+ if (!reason.StartsWith("blank", StringComparison.Ordinal))
+ {
+ staged.Warnings.Add($"Row {r + 1}: skipped ({reason}).");
+ }
+
+ continue;
+ }
+
+ if (!TryGetDate(row, profile, out var time))
+ {
+ staged.SkippedRows++;
+ staged.Warnings.Add($"Row {r + 1}: unparseable date, skipped.");
+ continue;
+ }
+
+ foreach (var column in profile.Columns)
+ {
+ StageColumn(column, row, time, r, staged, previousByMeter);
+ }
+ }
+
+ return staged;
+ }
+
+ private static void StageColumn(ColumnMapping column, IReadOnlyList row, DateTimeOffset time,
+ int rowIndex, StagedImport staged, Dictionary previousByMeter)
+ {
+ if (column.Role == MappingRole.Ignore || column.Index >= row.Count)
+ {
+ return;
+ }
+
+ var cell = row[column.Index];
+ if (string.IsNullOrWhiteSpace(cell))
+ {
+ return;
+ }
+
+ switch (column.Role)
+ {
+ case MappingRole.Reading:
+ StageReading(column, row, cell, time, rowIndex, staged, previousByMeter);
+ break;
+
+ case MappingRole.Delivery:
+ if (GermanNumber.TryParse(cell, out var litres) && litres > 0)
+ {
+ staged.Events.Add(new MeterEvent
+ {
+ MeterId = column.MeterId!.Value,
+ Time = time,
+ EventType = MeterEventType.Delivery,
+ Amount = litres,
+ Unit = column.Unit ?? "L",
+ });
+ }
+
+ break;
+
+ case MappingRole.TankLevel:
+ var split = ValueCell.Split(cell);
+ if (split is not null)
+ {
+ staged.Events.Add(new MeterEvent
+ {
+ MeterId = column.MeterId!.Value,
+ Time = time,
+ EventType = MeterEventType.TankLevel,
+ Amount = split.Value.Value,
+ Unit = column.Unit ?? split.Value.Unit,
+ });
+ }
+
+ break;
+
+ case MappingRole.ManualCost:
+ if (GermanMoney.TryParse(cell, out var amount, out var currency))
+ {
+ staged.ManualCosts.Add(new ManualCost
+ {
+ CategoryId = column.CategoryId,
+ PeriodStart = DateOnly.FromDateTime(time.UtcDateTime),
+ PeriodEnd = DateOnly.FromDateTime(time.UtcDateTime.AddMonths(1).AddDays(-1)),
+ Amount = (double)amount,
+ Currency = currency,
+ });
+ }
+
+ break;
+
+ case MappingRole.Ignore:
+ default:
+ break;
+ }
+ }
+
+ private static void StageReading(ColumnMapping column, IReadOnlyList row, string cell,
+ DateTimeOffset time, int rowIndex, StagedImport staged, Dictionary previousByMeter)
+ {
+ var split = ValueCell.Split(cell);
+ if (split is null)
+ {
+ return;
+ }
+
+ var meterId = column.MeterId!.Value;
+ var value = split.Value.Value;
+
+ if (column.Unit is not null && split.Value.Unit is not null
+ && !string.Equals(column.Unit, split.Value.Unit, StringComparison.OrdinalIgnoreCase))
+ {
+ staged.Warnings.Add(
+ $"Row {rowIndex + 1}: expected unit '{column.Unit}' but found '{split.Value.Unit}'.");
+ }
+
+ if (previousByMeter.TryGetValue(meterId, out var previous) && value < previous)
+ {
+ var override_ = ReadSwapOverride(column, row);
+ staged.Events.Add(new MeterEvent
+ {
+ MeterId = meterId,
+ Time = time,
+ EventType = MeterEventType.MeterSwap,
+ PrevValue = previous,
+ NewValue = value,
+ Amount = override_,
+ Notes = "Auto-detected register decrease (meter swap / reset candidate).",
+ });
+ staged.Warnings.Add(
+ $"Row {rowIndex + 1}: register for meter {meterId} dropped {previous}→{value}; " +
+ (override_ is null ? "swap event created (verify)." : $"swap consumption set to {override_}."));
+ }
+
+ staged.Readings.Add(new Reading
+ {
+ MeterId = meterId,
+ Time = time,
+ Value = value,
+ Quality = ReadingQuality.Imported,
+ });
+ previousByMeter[meterId] = value;
+ }
+
+ private static double? ReadSwapOverride(ColumnMapping column, IReadOnlyList row)
+ {
+ if (column.SwapConsumptionColumn is not { } source || source >= row.Count)
+ {
+ return null;
+ }
+
+ return GermanNumber.TryParse(row[source], out var value) ? value : null;
+ }
+
+ private static bool TryGetDate(IReadOnlyList row, MappingProfile profile, out DateTimeOffset time)
+ {
+ time = default;
+ if (profile.DateColumn >= row.Count)
+ {
+ return false;
+ }
+
+ var raw = row[profile.DateColumn];
+ DateOnly date;
+ var parsed = profile.DateKind switch
+ {
+ DateKind.MonthName => TryParseMonthAnchored(raw, profile.AnchorMonthsToEnd, out date),
+ DateKind.DayDotMonthYear => GermanDate.TryParseDay(raw, out date),
+ _ => ImportDate.TryResolve(raw, profile.AnchorMonthsToEnd, out date),
+ };
+
+ if (!parsed)
+ {
+ return false;
+ }
+
+ time = new DateTimeOffset(date.Year, date.Month, date.Day, 0, 0, 0, TimeSpan.Zero);
+ return true;
+ }
+
+ private static bool TryParseMonthAnchored(string? raw, bool anchorToEnd, out DateOnly date)
+ {
+ if (!GermanDate.TryParseMonth(raw, out var month))
+ {
+ date = default;
+ return false;
+ }
+
+ date = anchorToEnd
+ ? new DateOnly(month.Year, month.Month, DateTime.DaysInMonth(month.Year, month.Month))
+ : month;
+ return true;
+ }
+
+ internal static List ReadRows(TextReader reader)
+ {
+ var config = new CsvConfiguration(CultureInfo.InvariantCulture)
+ {
+ HasHeaderRecord = false,
+ DetectColumnCountChanges = false,
+ MissingFieldFound = null,
+ BadDataFound = null,
+ };
+
+ using var parser = new CsvParser(reader, config);
+ var rows = new List();
+ while (parser.Read())
+ {
+ rows.Add(parser.Record ?? []);
+ }
+
+ return rows;
+ }
+}
diff --git a/src/Infrastructure/Import/ImportDate.cs b/src/Infrastructure/Import/ImportDate.cs
new file mode 100644
index 0000000..37bbcb7
--- /dev/null
+++ b/src/Infrastructure/Import/ImportDate.cs
@@ -0,0 +1,31 @@
+using MeterVault.Core.Parsing;
+
+namespace MeterVault.Infrastructure.Import;
+
+///
+/// Resolves a cell to a date for import. Day-dated rows keep their day; month-name rows anchor to
+/// the first or last day of the month. Month-end anchoring matters when a sheet interleaves
+/// day-dated event rows with monthly rows (heating oil): a "Dezember 2022" snapshot must sort
+/// after a "05.12.2022" delivery reading, not before it.
+///
+public static class ImportDate
+{
+ public static bool TryResolve(string? raw, bool anchorMonthsToEnd, out DateOnly date)
+ {
+ if (GermanDate.TryParseDay(raw, out date))
+ {
+ return true;
+ }
+
+ if (GermanDate.TryParseMonth(raw, out var month))
+ {
+ date = anchorMonthsToEnd
+ ? new DateOnly(month.Year, month.Month, DateTime.DaysInMonth(month.Year, month.Month))
+ : month;
+ return true;
+ }
+
+ date = default;
+ return false;
+ }
+}
diff --git a/src/Infrastructure/Import/ImportService.cs b/src/Infrastructure/Import/ImportService.cs
new file mode 100644
index 0000000..223d998
--- /dev/null
+++ b/src/Infrastructure/Import/ImportService.cs
@@ -0,0 +1,96 @@
+using MeterVault.Infrastructure.Normalization;
+using MeterVault.Infrastructure.Persistence;
+using Microsoft.EntityFrameworkCore;
+
+namespace MeterVault.Infrastructure.Import;
+
+///
+/// Persists a as a revertible import_batch and recomputes the
+/// affected meters' consumption (SDD §6.3, FR-6). Every row it writes is tagged with the batch id
+/// so the whole import can be undone. Runs in a single transaction.
+///
+public sealed class ImportService(MeterVaultDbContext db, NormalizationService normalization)
+{
+ private readonly MeterVaultDbContext _db = db;
+ private readonly NormalizationService _normalization = normalization;
+
+ public async Task CommitAsync(
+ StagedImport staged, string? sourceName, string? mappingJson, CancellationToken cancellationToken = default)
+ {
+ ArgumentNullException.ThrowIfNull(staged);
+
+ await using var tx = await _db.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
+
+ var batch = new Core.Domain.ImportBatch
+ {
+ SourceName = sourceName,
+ Mapping = mappingJson,
+ RowCount = staged.TotalRows,
+ };
+ _db.ImportBatches.Add(batch);
+ await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
+
+ foreach (var reading in staged.Readings)
+ {
+ reading.ImportBatchId = batch.Id;
+ _db.Readings.Add(reading);
+ }
+
+ foreach (var meterEvent in staged.Events)
+ {
+ meterEvent.ImportBatchId = batch.Id;
+ _db.MeterEvents.Add(meterEvent);
+ }
+
+ foreach (var manualCost in staged.ManualCosts)
+ {
+ manualCost.ImportBatchId = batch.Id;
+ _db.ManualCosts.Add(manualCost);
+ }
+
+ await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
+
+ foreach (var meterId in AffectedMeters(staged))
+ {
+ await _normalization.RecomputeMeterAsync(meterId, batch.Id, cancellationToken).ConfigureAwait(false);
+ }
+
+ await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
+ await tx.CommitAsync(cancellationToken).ConfigureAwait(false);
+ return batch.Id;
+ }
+
+ public async Task RevertAsync(int batchId, CancellationToken cancellationToken = default)
+ {
+ await using var tx = await _db.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
+
+ var affected = await _db.Readings.Where(r => r.ImportBatchId == batchId).Select(r => r.MeterId)
+ .Union(_db.MeterEvents.Where(e => e.ImportBatchId == batchId).Select(e => e.MeterId))
+ .Distinct()
+ .ToListAsync(cancellationToken).ConfigureAwait(false);
+
+ await _db.Consumption.Where(c => c.ImportBatchId == batchId).ExecuteDeleteAsync(cancellationToken).ConfigureAwait(false);
+ await _db.Readings.Where(r => r.ImportBatchId == batchId).ExecuteDeleteAsync(cancellationToken).ConfigureAwait(false);
+ await _db.MeterEvents.Where(e => e.ImportBatchId == batchId).ExecuteDeleteAsync(cancellationToken).ConfigureAwait(false);
+ await _db.ManualCosts.Where(m => m.ImportBatchId == batchId).ExecuteDeleteAsync(cancellationToken).ConfigureAwait(false);
+
+ foreach (var meterId in affected)
+ {
+ await _normalization.RecomputeMeterAsync(meterId, null, cancellationToken).ConfigureAwait(false);
+ }
+
+ var batch = await _db.ImportBatches.FirstOrDefaultAsync(b => b.Id == batchId, cancellationToken).ConfigureAwait(false);
+ if (batch is not null)
+ {
+ batch.RevertedAt = DateTimeOffset.UtcNow;
+ }
+
+ await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
+ await tx.CommitAsync(cancellationToken).ConfigureAwait(false);
+ }
+
+ private static IEnumerable AffectedMeters(StagedImport staged) =>
+ staged.Readings.Select(r => r.MeterId)
+ .Concat(staged.Events.Select(e => e.MeterId))
+ .Distinct();
+}
diff --git a/src/Infrastructure/Import/MappingProfile.cs b/src/Infrastructure/Import/MappingProfile.cs
new file mode 100644
index 0000000..3520df1
--- /dev/null
+++ b/src/Infrastructure/Import/MappingProfile.cs
@@ -0,0 +1,89 @@
+namespace MeterVault.Infrastructure.Import;
+
+/// How a mapped column's cells are interpreted.
+public enum MappingRole
+{
+ /// Column carries derived/oracle values the importer ignores.
+ Ignore,
+
+ /// Cumulative/generation register or runtime hours → a reading.
+ Reading,
+
+ /// Delivery quantity → a delivery meter event.
+ Delivery,
+
+ /// Physical tank level (cm or L) → a tank_level meter event.
+ TankLevel,
+
+ /// A pre-computed monthly cost → a manual_cost row for a category.
+ ManualCost,
+}
+
+/// How the date column is parsed.
+public enum DateKind
+{
+ Auto,
+ MonthName,
+ DayDotMonthYear,
+}
+
+/// Maps a single spreadsheet column (by index) to a meter/category and a role.
+public sealed class ColumnMapping
+{
+ public required int Index { get; init; }
+
+ public MappingRole Role { get; init; }
+
+ /// Target meter for Reading/Delivery/TankLevel roles.
+ public int? MeterId { get; init; }
+
+ /// Target category for ManualCost roles.
+ public int? CategoryId { get; init; }
+
+ /// Expected unit; validated against the cell's unit suffix when present.
+ public string? Unit { get; init; }
+
+ ///
+ /// For a cumulative Reading column that can swap: the column whose value supplies the
+ /// swap-month consumption override (the sheet's own consumption column). SDD "spreadsheet wins".
+ ///
+ public int? SwapConsumptionColumn { get; init; }
+}
+
+///
+/// A declarative description of one spreadsheet shape (SDD FR-6, Appendix A). The four reference
+/// sheets ship as built-in profiles; the import wizard produces more. Columns are referenced by
+/// index because the sheets pack blank spacer and duplicated derived headers side by side.
+///
+public sealed class MappingProfile
+{
+ public static readonly IReadOnlyList DefaultSkipMarkers =
+ ["Total", "Heute", "Seitbeginn", "Seit ", "Gesammt"];
+
+ public required string Name { get; init; }
+
+ /// 0-based index of the data header row (Heizöl's real header is not row 0).
+ public int HeaderRowIndex { get; init; }
+
+ /// 0-based index of the first data row.
+ public int FirstDataRowIndex { get; init; }
+
+ public required int DateColumn { get; init; }
+
+ public DateKind DateKind { get; init; } = DateKind.Auto;
+
+ public IReadOnlyList Columns { get; init; } = [];
+
+ public IReadOnlyList SkipRowMarkers { get; init; } = DefaultSkipMarkers;
+
+ public bool SkipAllZeroRows { get; init; } = true;
+
+ /// Emit a meter swap when a cumulative register drops without an explaining event.
+ public bool DetectCumulativeSwaps { get; init; }
+
+ ///
+ /// Anchor month-name rows to the last day of the month instead of the first. Needed when
+ /// day-dated event rows are interleaved with monthly rows (heating oil).
+ ///
+ public bool AnchorMonthsToEnd { get; init; }
+}
diff --git a/src/Infrastructure/Import/ReferenceProfiles.cs b/src/Infrastructure/Import/ReferenceProfiles.cs
new file mode 100644
index 0000000..7380e4b
--- /dev/null
+++ b/src/Infrastructure/Import/ReferenceProfiles.cs
@@ -0,0 +1,96 @@
+namespace MeterVault.Infrastructure.Import;
+
+///
+/// Built-in mapping profiles for the four reference Energiebilanz sheets (SDD §6.3).
+/// They ship as example imports and as the golden reconciliation fixtures. Meter and category ids
+/// are fixed conventions for the reference data; the import wizard maps to real ids at runtime.
+///
+public static class ReferenceProfiles
+{
+ // Reference meter ids.
+ public const int Haus = 1;
+ public const int Netz = 2;
+ public const int Auto = 3;
+ public const int Solar1 = 4;
+ public const int Solar2 = 5;
+ public const int Wasser = 10;
+ public const int OilTank = 20;
+ public const int Burner = 21;
+
+ // Reference category ids (match DatabaseSeeder order).
+ public const int CategoryHeizung = 1;
+ public const int CategoryStrom = 2;
+ public const int CategoryWasser = 3;
+ public const int CategoryPool = 4;
+
+ /// Linear heating-oil tank calibration: 7000 L / 150 cm ≈ 46.667 L/cm.
+ public const double OilLitresPerCm = 7000d / 150d;
+
+ public static MappingProfile Electricity() => new()
+ {
+ Name = "Energiebilanz — Strom",
+ DateColumn = 0,
+ DateKind = DateKind.MonthName,
+ FirstDataRowIndex = 1,
+ Columns =
+ [
+ new ColumnMapping { Index = 1, Role = MappingRole.Reading, MeterId = Haus, Unit = "kWh" },
+ new ColumnMapping { Index = 2, Role = MappingRole.Reading, MeterId = Netz, Unit = "kWh" },
+ // Index 3 is a blank spacer column — left unmapped (Ignore).
+ new ColumnMapping { Index = 4, Role = MappingRole.Reading, MeterId = Auto, Unit = "kWh" },
+ new ColumnMapping { Index = 5, Role = MappingRole.Reading, MeterId = Solar1, Unit = "kWh" },
+ new ColumnMapping { Index = 6, Role = MappingRole.Reading, MeterId = Solar2, Unit = "kWh" },
+ ],
+ };
+
+ public static MappingProfile Water() => new()
+ {
+ Name = "Energiebilanz — Wasser",
+ DateColumn = 0,
+ DateKind = DateKind.MonthName,
+ FirstDataRowIndex = 1,
+ DetectCumulativeSwaps = true,
+ Columns =
+ [
+ new ColumnMapping
+ {
+ Index = 1,
+ Role = MappingRole.Reading,
+ MeterId = Wasser,
+ Unit = "m3",
+ SwapConsumptionColumn = 2, // Wasserverbrauch supplies the swap-month consumption.
+ },
+ ],
+ };
+
+ public static MappingProfile HeatingOil() => new()
+ {
+ Name = "Energiebilanz — Heizöl",
+ DateColumn = 0,
+ DateKind = DateKind.Auto, // early rows DD.MM.YYYY, later rows month names.
+ AnchorMonthsToEnd = true, // a monthly snapshot sorts after same-month day-dated readings.
+ HeaderRowIndex = 3,
+ FirstDataRowIndex = 4,
+ Columns =
+ [
+ new ColumnMapping { Index = 1, Role = MappingRole.Reading, MeterId = Burner, Unit = "h" },
+ new ColumnMapping { Index = 4, Role = MappingRole.TankLevel, MeterId = OilTank, Unit = "cm" },
+ new ColumnMapping { Index = 7, Role = MappingRole.Delivery, MeterId = OilTank, Unit = "L" },
+ ],
+ };
+
+ public static MappingProfile Costs() => new()
+ {
+ Name = "Energiebilanz — Kosten",
+ DateColumn = 0,
+ DateKind = DateKind.MonthName,
+ FirstDataRowIndex = 1,
+ Columns =
+ [
+ new ColumnMapping { Index = 3, Role = MappingRole.ManualCost, CategoryId = CategoryHeizung },
+ new ColumnMapping { Index = 4, Role = MappingRole.ManualCost, CategoryId = CategoryStrom },
+ new ColumnMapping { Index = 5, Role = MappingRole.ManualCost, CategoryId = CategoryWasser },
+ new ColumnMapping { Index = 6, Role = MappingRole.ManualCost, CategoryId = CategoryPool },
+ ],
+ };
+}
diff --git a/src/Infrastructure/Import/RowClassifier.cs b/src/Infrastructure/Import/RowClassifier.cs
new file mode 100644
index 0000000..b1e3232
--- /dev/null
+++ b/src/Infrastructure/Import/RowClassifier.cs
@@ -0,0 +1,73 @@
+using MeterVault.Core.Parsing;
+
+namespace MeterVault.Infrastructure.Import;
+
+///
+/// Decides which CSV rows to ingest (SDD Appendix A): skips blank/separator rows, inline summary
+/// rows (Total / Heute / Seitbeginn / Seit … / Gesammt), and all-zero future placeholders.
+///
+public static class RowClassifier
+{
+ public static bool ShouldSkip(IReadOnlyList row, MappingProfile profile, out string reason)
+ {
+ ArgumentNullException.ThrowIfNull(row);
+ ArgumentNullException.ThrowIfNull(profile);
+
+ if (row.Count == 0 || row.All(string.IsNullOrWhiteSpace))
+ {
+ reason = "blank/separator row";
+ return true;
+ }
+
+ var first = (row.Count > 0 ? row[0] : string.Empty).Trim();
+ foreach (var marker in profile.SkipRowMarkers)
+ {
+ if (first.StartsWith(marker, StringComparison.OrdinalIgnoreCase))
+ {
+ reason = $"summary row '{first}'";
+ return true;
+ }
+ }
+
+ if (profile.SkipAllZeroRows && IsAllZeroData(row, profile))
+ {
+ reason = "all-zero placeholder";
+ return true;
+ }
+
+ reason = string.Empty;
+ return false;
+ }
+
+ private static bool IsAllZeroData(IReadOnlyList row, MappingProfile profile)
+ {
+ var sawValueColumn = false;
+ foreach (var column in profile.Columns)
+ {
+ if (column.Role == MappingRole.Ignore || column.Index >= row.Count)
+ {
+ continue;
+ }
+
+ sawValueColumn = true;
+ var cell = row[column.Index];
+ if (string.IsNullOrWhiteSpace(cell))
+ {
+ continue;
+ }
+
+ // Any non-zero numeric in a value column means the row carries data.
+ if (GermanNumber.TryParse(cell, out var value) && value != 0)
+ {
+ return false;
+ }
+
+ if (!GermanNumber.TryParse(cell, out _) && GermanMoney.TryParse(cell, out var money, out _) && money != 0)
+ {
+ return false;
+ }
+ }
+
+ return sawValueColumn;
+ }
+}
diff --git a/src/Infrastructure/Import/StagedImport.cs b/src/Infrastructure/Import/StagedImport.cs
new file mode 100644
index 0000000..6f1fae4
--- /dev/null
+++ b/src/Infrastructure/Import/StagedImport.cs
@@ -0,0 +1,23 @@
+using MeterVault.Core.Domain;
+
+namespace MeterVault.Infrastructure.Import;
+
+///
+/// The result of staging a CSV against a profile: the readings/events/manual-costs it would
+/// create, plus warnings (detected swaps, anomalies, unit mismatches) and a skipped-row count.
+/// Nothing is persisted until commits it.
+///
+public sealed class StagedImport
+{
+ public List Readings { get; } = [];
+
+ public List Events { get; } = [];
+
+ public List ManualCosts { get; } = [];
+
+ public List Warnings { get; } = [];
+
+ public int SkippedRows { get; set; }
+
+ public int TotalRows => Readings.Count + Events.Count + ManualCosts.Count;
+}
diff --git a/src/Infrastructure/Normalization/MeterConfigFactory.cs b/src/Infrastructure/Normalization/MeterConfigFactory.cs
new file mode 100644
index 0000000..15ce6e3
--- /dev/null
+++ b/src/Infrastructure/Normalization/MeterConfigFactory.cs
@@ -0,0 +1,88 @@
+using System.Text.Json;
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+
+namespace MeterVault.Infrastructure.Normalization;
+
+///
+/// Projects a persisted (and optional ) into the
+/// infrastructure-free the normalization engine consumes.
+///
+public static class MeterConfigFactory
+{
+ public static MeterConfig FromMeter(Meter meter, Tank? tank)
+ {
+ ArgumentNullException.ThrowIfNull(meter);
+
+ return new MeterConfig
+ {
+ MeterId = meter.Id,
+ Mode = meter.Mode,
+ Unit = meter.Unit,
+ InitialBaseline = meter.InitialBaseline,
+ Tank = tank is null ? null : new TankConfig
+ {
+ Capacity = tank.Capacity,
+ RateMode = tank.RateMode,
+ FixedRate = tank.FixedRate,
+ Calibration = ParseCalibration(tank.Calibration),
+ },
+ Virtual = ParseVirtual(meter),
+ };
+ }
+
+ private static CalibrationCurve? ParseCalibration(string? json)
+ {
+ if (string.IsNullOrWhiteSpace(json))
+ {
+ return null;
+ }
+
+ try
+ {
+ using var doc = JsonDocument.Parse(json);
+ var root = doc.RootElement;
+ if (!root.TryGetProperty("volumePerUnit", out var perUnit))
+ {
+ return null;
+ }
+
+ var offset = root.TryGetProperty("offset", out var off) ? off.GetDouble() : 0d;
+ return new CalibrationCurve(perUnit.GetDouble(), offset);
+ }
+ catch (JsonException)
+ {
+ return null;
+ }
+ }
+
+ private static VirtualSpec? ParseVirtual(Meter meter)
+ {
+ if (meter.Mode != MeterMode.Virtual || string.IsNullOrWhiteSpace(meter.Meta))
+ {
+ return null;
+ }
+
+ try
+ {
+ using var doc = JsonDocument.Parse(meter.Meta);
+ var root = doc.RootElement;
+ if (!root.TryGetProperty("expression", out var expression))
+ {
+ return null;
+ }
+
+ var referenced = new List();
+ if (root.TryGetProperty("referencedMeterIds", out var ids) && ids.ValueKind == JsonValueKind.Array)
+ {
+ referenced.AddRange(ids.EnumerateArray().Select(e => e.GetInt32()));
+ }
+
+ return new VirtualSpec { Expression = expression.GetString() ?? string.Empty, ReferencedMeterIds = referenced };
+ }
+ catch (JsonException)
+ {
+ return null;
+ }
+ }
+}
diff --git a/src/Infrastructure/Normalization/NormalizationService.cs b/src/Infrastructure/Normalization/NormalizationService.cs
new file mode 100644
index 0000000..d71f037
--- /dev/null
+++ b/src/Infrastructure/Normalization/NormalizationService.cs
@@ -0,0 +1,55 @@
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+using MeterVault.Infrastructure.Persistence;
+using Microsoft.EntityFrameworkCore;
+
+namespace MeterVault.Infrastructure.Normalization;
+
+///
+/// Bridges persisted readings/events to the pure engine: loads a meter's inputs, recomputes its
+/// consumption wholesale (consumption is a pure function of readings + events), and replaces the
+/// stored rows. Virtual meters are skipped here — they are computed on read (SDD §14.1).
+///
+public sealed class NormalizationService(MeterVaultDbContext db, INormalizationEngine engine)
+{
+ private readonly MeterVaultDbContext _db = db;
+ private readonly INormalizationEngine _engine = engine;
+
+ ///
+ /// Recomputes and replaces the consumption series for one meter from all its current readings
+ /// and events. Tags new rows with for provenance. Does not save.
+ ///
+ public async Task RecomputeMeterAsync(int meterId, int? batchId, CancellationToken cancellationToken = default)
+ {
+ var meter = await _db.Meters.FirstOrDefaultAsync(m => m.Id == meterId, cancellationToken).ConfigureAwait(false);
+ if (meter is null || meter.Mode == MeterMode.Virtual)
+ {
+ return;
+ }
+
+ var tank = await _db.Tanks.FirstOrDefaultAsync(t => t.MeterId == meterId, cancellationToken).ConfigureAwait(false);
+ var config = MeterConfigFactory.FromMeter(meter, tank);
+
+ var readings = await _db.Readings
+ .Where(r => r.MeterId == meterId)
+ .OrderBy(r => r.Time)
+ .ToListAsync(cancellationToken).ConfigureAwait(false);
+
+ var events = await _db.MeterEvents
+ .Where(e => e.MeterId == meterId)
+ .OrderBy(e => e.Time)
+ .ToListAsync(cancellationToken).ConfigureAwait(false);
+
+ var context = new NormalizationContext { Meter = config, Readings = readings, Events = events };
+ var consumption = _engine.Normalize(context);
+
+ await _db.Consumption.Where(c => c.MeterId == meterId)
+ .ExecuteDeleteAsync(cancellationToken).ConfigureAwait(false);
+
+ foreach (var row in consumption)
+ {
+ row.ImportBatchId = batchId;
+ _db.Consumption.Add(row);
+ }
+ }
+}
diff --git a/tests/Core.Tests/GermanParsingTests.cs b/tests/Core.Tests/GermanParsingTests.cs
new file mode 100644
index 0000000..ae6c644
--- /dev/null
+++ b/tests/Core.Tests/GermanParsingTests.cs
@@ -0,0 +1,81 @@
+using MeterVault.Core.Parsing;
+
+namespace MeterVault.Core.Tests;
+
+public sealed class GermanParsingTests
+{
+ [Theory]
+ [InlineData("411kWh", 411)]
+ [InlineData("2.940,19", 2940.19)]
+ [InlineData("180,8244706", 180.8244706)]
+ [InlineData("-90kWh", -90)]
+ [InlineData("0,31", 0.31)]
+ [InlineData("49", 49)]
+ [InlineData("2287", 2287)]
+ [InlineData("1.295,19 €", 1295.19)]
+ [InlineData("5,98", 5.98)]
+ [InlineData("0kWh", 0)]
+ public void GermanNumber_parses_dialect(string raw, double expected)
+ {
+ Assert.True(GermanNumber.TryParse(raw, out var value));
+ Assert.Equal(expected, value, 6);
+ }
+
+ [Theory]
+ [InlineData("")]
+ [InlineData(" ")]
+ [InlineData("kWh")]
+ [InlineData("abc")]
+ public void GermanNumber_rejects_non_numeric(string raw) =>
+ Assert.False(GermanNumber.TryParse(raw, out _));
+
+ [Theory]
+ [InlineData("120,00 €", 120.00, "EUR")]
+ [InlineData("2.940,19 €", 2940.19, "EUR")]
+ [InlineData("70 €", 70, "EUR")]
+ [InlineData("-0,80 €", -0.80, "EUR")]
+ public void GermanMoney_parses_currency(string raw, decimal expected, string currency)
+ {
+ Assert.True(GermanMoney.TryParse(raw, out var amount, out var cur));
+ Assert.Equal(expected, amount);
+ Assert.Equal(currency, cur);
+ }
+
+ [Theory]
+ [InlineData("2287L", 2287, "L")]
+ [InlineData("411kWh", 411, "kWh")]
+ [InlineData("49", 49, null)]
+ [InlineData("35 cm", 35, "cm")]
+ public void ValueCell_splits_value_and_unit(string raw, double value, string? unit)
+ {
+ var result = ValueCell.Split(raw);
+ Assert.NotNull(result);
+ Assert.Equal(value, result!.Value.Value, 6);
+ Assert.Equal(unit, result.Value.Unit);
+ }
+
+ [Theory]
+ [InlineData("September 2022", 2022, 9)]
+ [InlineData("März 2023", 2023, 3)]
+ [InlineData("Dezember 2026", 2026, 12)]
+ public void GermanDate_parses_month_tables(string raw, int year, int month)
+ {
+ Assert.True(GermanDate.TryParseMonth(raw, out var date));
+ Assert.Equal(new DateOnly(year, month, 1), date);
+ }
+
+ [Theory]
+ [InlineData("10.06.1997", 1997, 6, 10)]
+ [InlineData("13.07.2026", 2026, 7, 13)]
+ public void GermanDate_parses_event_rows(string raw, int year, int month, int day)
+ {
+ Assert.True(GermanDate.TryParseDay(raw, out var date));
+ Assert.Equal(new DateOnly(year, month, day), date);
+ }
+
+ [Theory]
+ [InlineData("10.06.1997")]
+ [InlineData("September 2022")]
+ public void GermanDate_auto_detects_both_shapes(string raw) =>
+ Assert.True(GermanDate.TryParse(raw, out _));
+}
diff --git a/tests/Integration.Tests/ImportRoundTripTests.cs b/tests/Integration.Tests/ImportRoundTripTests.cs
new file mode 100644
index 0000000..f1f4171
--- /dev/null
+++ b/tests/Integration.Tests/ImportRoundTripTests.cs
@@ -0,0 +1,88 @@
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+using MeterVault.Infrastructure.Import;
+using MeterVault.Infrastructure.Normalization;
+using MeterVault.Infrastructure.Persistence;
+using Microsoft.EntityFrameworkCore;
+using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
+
+namespace MeterVault.Integration.Tests;
+
+///
+/// End-to-end persistence: stage the water CSV → commit as a batch → normalized consumption lands
+/// in the hypertable → revert removes everything and rebases consumption (SDD §6.3, FR-6).
+///
+[Collection("Timescale")]
+public sealed class ImportRoundTripTests(TimescaleFixture fx)
+{
+ [Fact]
+ public async Task Commit_persists_consumption_and_revert_removes_it()
+ {
+ await using var db = fx.CreateContext();
+ await DatabaseSeeder.SeedAsync(db);
+
+ var waterType = await db.EnergyTypes.FirstAsync(t => t.Key == "water");
+ var meter = new Meter
+ {
+ Name = "Zähler Wasser (round-trip test)",
+ EnergyTypeId = waterType.Id,
+ Mode = MeterMode.CumulativeCounter,
+ Unit = "m3",
+ InitialBaseline = 820,
+ };
+ db.Meters.Add(meter);
+ await db.SaveChangesAsync();
+
+ var profile = new MappingProfile
+ {
+ Name = "test-water",
+ DateColumn = 0,
+ DateKind = DateKind.MonthName,
+ FirstDataRowIndex = 1,
+ DetectCumulativeSwaps = true,
+ Columns =
+ [
+ new ColumnMapping
+ {
+ Index = 1,
+ Role = MappingRole.Reading,
+ MeterId = meter.Id,
+ Unit = "m3",
+ SwapConsumptionColumn = 2,
+ },
+ ],
+ };
+
+ StagedImport staged;
+ using (var reader = new StreamReader(FixturePath(Water)))
+ {
+ staged = new CsvImporter().Stage(profile, reader);
+ }
+
+ var normalization = new NormalizationService(db, NormalizationEngine.CreateDefault());
+ var service = new ImportService(db, normalization);
+
+ // Commit.
+ var batchId = await service.CommitAsync(staged, sourceName: "Wasser.csv", mappingJson: null);
+
+ Assert.True(await db.Readings.AnyAsync(r => r.MeterId == meter.Id));
+ Assert.True(await db.MeterEvents.AnyAsync(e => e.MeterId == meter.Id && e.EventType == MeterEventType.MeterSwap));
+
+ var maerz = await db.Consumption.SingleAsync(c =>
+ c.MeterId == meter.Id && c.Time == new DateTimeOffset(2023, 3, 1, 0, 0, 0, TimeSpan.Zero));
+ Assert.Equal(12d, maerz.Amount, 3);
+
+ // Revert.
+ await service.RevertAsync(batchId);
+
+ Assert.False(await db.Readings.AnyAsync(r => r.MeterId == meter.Id));
+ Assert.False(await db.MeterEvents.AnyAsync(e => e.MeterId == meter.Id));
+ Assert.False(await db.Consumption.AnyAsync(c => c.MeterId == meter.Id));
+ var batch = await db.ImportBatches.SingleAsync(b => b.Id == batchId);
+ Assert.NotNull(batch.RevertedAt);
+
+ // Cleanup so the shared container stays tidy for other tests. ExecuteDelete bypasses the
+ // change tracker (which holds stale entries after the revert's set-based deletes).
+ await db.Meters.Where(m => m.Id == meter.Id).ExecuteDeleteAsync();
+ }
+}
diff --git a/tests/Integration.Tests/MeterVault.Integration.Tests.csproj b/tests/Integration.Tests/MeterVault.Integration.Tests.csproj
index cb325af..fb5304e 100644
--- a/tests/Integration.Tests/MeterVault.Integration.Tests.csproj
+++ b/tests/Integration.Tests/MeterVault.Integration.Tests.csproj
@@ -14,6 +14,11 @@
+
+
+
+
+
diff --git a/tests/Integration.Tests/Reconciliation/CostsReconciliationTests.cs b/tests/Integration.Tests/Reconciliation/CostsReconciliationTests.cs
new file mode 100644
index 0000000..202af12
--- /dev/null
+++ b/tests/Integration.Tests/Reconciliation/CostsReconciliationTests.cs
@@ -0,0 +1,41 @@
+using MeterVault.Infrastructure.Import;
+using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
+
+namespace MeterVault.Integration.Tests.Reconciliation;
+
+///
+/// The Kosten sheet is pre-computed category costs (SDD §2.1). The importer stages one manual
+/// cost per category column; their monthly sum must equal the sheet's total Kosten column, and
+/// the meterless "Pool Betrieb" column becomes manual costs with no meter.
+///
+public sealed class CostsReconciliationTests
+{
+ [Fact]
+ public void Category_costs_sum_to_the_monthly_total()
+ {
+ var staged = Stage(ReferenceProfiles.Costs(), Costs);
+
+ var computed = staged.ManualCosts
+ .GroupBy(c => new DateOnly(c.PeriodStart.Year, c.PeriodStart.Month, 1))
+ .ToDictionary(g => g.Key, g => g.Sum(c => c.Amount));
+
+ var oracle = OracleByMonth(ReadRows(Costs), dateColumn: 0, valueColumn: 2, firstDataRow: 1);
+
+ AssertReconciles(computed, oracle, tolerance: 0.02, "category totals", minMatches: 20);
+ }
+
+ [Fact]
+ public void Manual_costs_are_staged_meterless_by_category()
+ {
+ var staged = Stage(ReferenceProfiles.Costs(), Costs);
+
+ Assert.NotEmpty(staged.ManualCosts);
+ Assert.All(staged.ManualCosts, c => Assert.Null(c.MeterId));
+
+ // Wasser Dez 2022 = 70,00 € (SDD §2.1: categories decoupled from meters).
+ var wasserDec = staged.ManualCosts.Single(c =>
+ c.CategoryId == ReferenceProfiles.CategoryWasser
+ && c.PeriodStart == new DateOnly(2022, 12, 1));
+ Assert.Equal(70d, wasserDec.Amount, 2);
+ }
+}
diff --git a/tests/Integration.Tests/Reconciliation/ElectricityReconciliationTests.cs b/tests/Integration.Tests/Reconciliation/ElectricityReconciliationTests.cs
new file mode 100644
index 0000000..9f7a94a
--- /dev/null
+++ b/tests/Integration.Tests/Reconciliation/ElectricityReconciliationTests.cs
@@ -0,0 +1,72 @@
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+using MeterVault.Infrastructure.Import;
+using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
+
+namespace MeterVault.Integration.Tests.Reconciliation;
+
+///
+/// Reconciles the five electricity meters and a derived virtual meter against the Strom sheet's
+/// own Verbrauch / Netz Einsparung columns (SDD §2.2). Runs without a database.
+///
+public sealed class ElectricityReconciliationTests
+{
+ private static MeterConfig Cumulative(int id) =>
+ new() { MeterId = id, Mode = MeterMode.CumulativeCounter, Unit = "kWh" };
+
+ private static MeterConfig Generation(int id) =>
+ new() { MeterId = id, Mode = MeterMode.GenerationCounter, Unit = "kWh" };
+
+ [Theory]
+ [InlineData(ReferenceProfiles.Haus, 7)] // Haus Verbrauch
+ [InlineData(ReferenceProfiles.Netz, 8)] // Netz Verbrauch
+ [InlineData(ReferenceProfiles.Auto, 9)] // Auto Verbrauch
+ [InlineData(ReferenceProfiles.Solar1, 10)] // Solar Erzeugung 1
+ [InlineData(ReferenceProfiles.Solar2, 11)] // Solar Erzeugung 2
+ public void Meter_consumption_matches_the_sheet(int meterId, int oracleColumn)
+ {
+ var staged = Stage(ReferenceProfiles.Electricity(), Electricity);
+ var config = meterId is ReferenceProfiles.Solar1 or ReferenceProfiles.Solar2
+ ? Generation(meterId)
+ : Cumulative(meterId);
+
+ var computed = ByMonth(Normalize(staged, config));
+ var oracle = OracleByMonth(ReadRows(Electricity), dateColumn: 0, valueColumn: oracleColumn, firstDataRow: 1);
+
+ AssertReconciles(computed, oracle, tolerance: 1.0, $"meter {meterId}", minMatches: 20);
+ }
+
+ [Fact]
+ public void Netz_einsparung_virtual_matches_the_sheet()
+ {
+ var staged = Stage(ReferenceProfiles.Electricity(), Electricity);
+ var haus = Normalize(staged, Cumulative(ReferenceProfiles.Haus));
+ var netz = Normalize(staged, Cumulative(ReferenceProfiles.Netz));
+
+ var engine = NormalizationEngine.CreateDefault();
+ var virtualContext = new NormalizationContext
+ {
+ Meter = new MeterConfig
+ {
+ MeterId = 100,
+ Mode = MeterMode.Virtual,
+ Unit = "kWh",
+ Virtual = new VirtualSpec
+ {
+ Expression = $"m{ReferenceProfiles.Haus} - m{ReferenceProfiles.Netz}",
+ ReferencedMeterIds = [ReferenceProfiles.Haus, ReferenceProfiles.Netz],
+ },
+ },
+ ReferencedSeries = new Dictionary>
+ {
+ [ReferenceProfiles.Haus] = haus,
+ [ReferenceProfiles.Netz] = netz,
+ },
+ };
+
+ var computed = ByMonth(engine.Normalize(virtualContext));
+ var oracle = OracleByMonth(ReadRows(Electricity), dateColumn: 0, valueColumn: 13, firstDataRow: 1); // Netz Einsparung
+
+ AssertReconciles(computed, oracle, tolerance: 1.0, "Netz Einsparung", minMatches: 20);
+ }
+}
diff --git a/tests/Integration.Tests/Reconciliation/OilReconciliationTests.cs b/tests/Integration.Tests/Reconciliation/OilReconciliationTests.cs
new file mode 100644
index 0000000..01ebe03
--- /dev/null
+++ b/tests/Integration.Tests/Reconciliation/OilReconciliationTests.cs
@@ -0,0 +1,56 @@
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+using MeterVault.Infrastructure.Import;
+using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
+
+namespace MeterVault.Integration.Tests.Reconciliation;
+
+///
+/// Reconciles the heating-oil tank against the Heizöl sheet (SDD §2.4). The sheet's
+/// Differenz Tank (col 9) is the negated actual consumption, already netting deliveries —
+/// exactly what the consumable-balance normalizer computes (prevLevel + deliveries − currLevel).
+/// The burner runtime meter's Δhours is reconciled against Differenz Betrieb (col 2).
+///
+public sealed class OilReconciliationTests
+{
+ [Fact]
+ public void Tank_consumption_matches_differenz_tank_including_deliveries()
+ {
+ var staged = Stage(ReferenceProfiles.HeatingOil(), Oil);
+ var config = new MeterConfig
+ {
+ MeterId = ReferenceProfiles.OilTank,
+ Mode = MeterMode.ConsumableBalance,
+ Unit = "L",
+ Tank = new TankConfig
+ {
+ Capacity = 7000,
+ Calibration = new CalibrationCurve(ReferenceProfiles.OilLitresPerCm),
+ },
+ };
+
+ var computed = ByDate(Normalize(staged, config));
+ // Differenz Tank is negative consumption → negate to compare with positive draw.
+ var oracle = OracleByDate(ReadRows(Oil), dateColumn: 0, valueColumn: 9, firstDataRow: 4, v => -v);
+
+ // ±2 L covers cm→litre rounding on both ends of each interval.
+ AssertReconciles(computed, oracle, tolerance: 2.0, "oil tank", minMatches: 30);
+ }
+
+ [Fact]
+ public void Burner_delta_hours_match_differenz_betrieb()
+ {
+ var staged = Stage(ReferenceProfiles.HeatingOil(), Oil);
+ var config = new MeterConfig
+ {
+ MeterId = ReferenceProfiles.Burner,
+ Mode = MeterMode.RuntimeCounter,
+ Unit = "h",
+ };
+
+ var computed = ByDate(Normalize(staged, config));
+ var oracle = OracleByDate(ReadRows(Oil), dateColumn: 0, valueColumn: 2, firstDataRow: 4);
+
+ AssertReconciles(computed, oracle, tolerance: 0.5, "burner hours", minMatches: 30);
+ }
+}
diff --git a/tests/Integration.Tests/Reconciliation/ReconciliationSupport.cs b/tests/Integration.Tests/Reconciliation/ReconciliationSupport.cs
new file mode 100644
index 0000000..a927cae
--- /dev/null
+++ b/tests/Integration.Tests/Reconciliation/ReconciliationSupport.cs
@@ -0,0 +1,146 @@
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+using MeterVault.Core.Parsing;
+using MeterVault.Infrastructure.Import;
+
+namespace MeterVault.Integration.Tests.Reconciliation;
+
+///
+/// Shared helpers for the golden-fixture reconciliation tests. The CSVs are self-oracling: the
+/// same file carries both the input (registers/levels/hours) and the expected output
+/// (Verbrauch / Differenz Tank / Kosten columns). We parse input → normalize → compare against
+/// the sheet's own columns (SDD §0.3, §13). No database is involved.
+///
+internal static class ReconciliationSupport
+{
+ // Fixture file names (linked into fixtures/ in the test output).
+ public const string Electricity = "Energiebilanz - Strom Verbrauch.csv";
+ public const string Water = "Energiebilanz - Wasser.csv";
+ public const string Oil = "Energiebilanz - Heizöl Verbrauch.csv";
+ public const string Costs = "Energiebilanz - Kosten.csv";
+
+ public static string FixturePath(string fileName) =>
+ Path.Combine(AppContext.BaseDirectory, "fixtures", fileName);
+
+ public static List ReadRows(string fileName)
+ {
+ using var reader = new StreamReader(FixturePath(fileName));
+ return CsvImporter.ReadRows(reader);
+ }
+
+ public static StagedImport Stage(MappingProfile profile, string fileName)
+ {
+ using var reader = new StreamReader(FixturePath(fileName));
+ return new CsvImporter().Stage(profile, reader);
+ }
+
+ /// Normalizes one meter from a staged import using the given config.
+ public static IReadOnlyList Normalize(StagedImport staged, MeterConfig config)
+ {
+ var engine = NormalizationEngine.CreateDefault();
+ var context = new NormalizationContext
+ {
+ Meter = config,
+ Readings = staged.Readings.Where(r => r.MeterId == config.MeterId).ToList(),
+ Events = staged.Events.Where(e => e.MeterId == config.MeterId).ToList(),
+ };
+
+ return engine.Normalize(context);
+ }
+
+ /// Extracts a sheet oracle column keyed by month, using German number parsing.
+ public static Dictionary OracleByMonth(
+ IReadOnlyList rows, int dateColumn, int valueColumn, int firstDataRow)
+ {
+ var result = new Dictionary();
+ for (var r = firstDataRow; r < rows.Count; r++)
+ {
+ var row = rows[r];
+ if (dateColumn >= row.Length || valueColumn >= row.Length)
+ {
+ continue;
+ }
+
+ if (!GermanDate.TryParse(row[dateColumn], out var date))
+ {
+ continue;
+ }
+
+ if (GermanNumber.TryParse(row[valueColumn], out var value))
+ {
+ result[new DateOnly(date.Year, date.Month, 1)] = value;
+ }
+ }
+
+ return result;
+ }
+
+ public static DateOnly MonthKey(DateTimeOffset time) => new(time.Year, time.Month, 1);
+
+ ///
+ /// Asserts every month present in both computed and oracle agrees within tolerance, and that
+ /// a meaningful number of months were actually compared (so an empty result can't pass).
+ ///
+ public static void AssertReconciles(
+ IReadOnlyDictionary computed,
+ IReadOnlyDictionary oracle,
+ double tolerance,
+ string label,
+ int minMatches)
+ {
+ var matched = 0;
+ foreach (var (month, expected) in oracle)
+ {
+ if (!computed.TryGetValue(month, out var actual))
+ {
+ continue;
+ }
+
+ matched++;
+ Assert.True(
+ Math.Abs(actual - expected) <= tolerance,
+ $"{label} {month:yyyy-MM}: computed {actual:0.##} vs sheet {expected:0.##} (tol {tolerance}).");
+ }
+
+ Assert.True(matched >= minMatches, $"{label}: only {matched} months reconciled (expected ≥ {minMatches}).");
+ }
+
+ public static Dictionary ByMonth(IReadOnlyList series) =>
+ series.GroupBy(c => MonthKey(c.Time)).ToDictionary(g => g.Key, g => g.Sum(c => c.Amount));
+
+ /// Consumption keyed by exact reading date (oil rows are event-dated, sometimes two per month).
+ public static Dictionary ByDate(IReadOnlyList series) =>
+ series.GroupBy(c => DateOnly.FromDateTime(c.Time.UtcDateTime))
+ .ToDictionary(g => g.Key, g => g.Sum(c => c.Amount));
+
+ ///
+ /// Extracts a sheet oracle column keyed by exact date, optionally transformed. Uses the same
+ /// month-end anchoring as the importer so day-dated and month-dated rows align.
+ ///
+ public static Dictionary OracleByDate(
+ IReadOnlyList rows, int dateColumn, int valueColumn, int firstDataRow,
+ Func? transform = null, bool anchorMonthsToEnd = true)
+ {
+ var result = new Dictionary();
+ for (var r = firstDataRow; r < rows.Count; r++)
+ {
+ var row = rows[r];
+ if (dateColumn >= row.Length || valueColumn >= row.Length)
+ {
+ continue;
+ }
+
+ if (!ImportDate.TryResolve(row[dateColumn], anchorMonthsToEnd, out var date))
+ {
+ continue;
+ }
+
+ if (GermanNumber.TryParse(row[valueColumn], out var value))
+ {
+ result[date] = transform is null ? value : transform(value);
+ }
+ }
+
+ return result;
+ }
+}
diff --git a/tests/Integration.Tests/Reconciliation/WaterReconciliationTests.cs b/tests/Integration.Tests/Reconciliation/WaterReconciliationTests.cs
new file mode 100644
index 0000000..2ef6e43
--- /dev/null
+++ b/tests/Integration.Tests/Reconciliation/WaterReconciliationTests.cs
@@ -0,0 +1,39 @@
+using MeterVault.Core.Domain;
+using MeterVault.Core.Normalization;
+using MeterVault.Infrastructure.Import;
+using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
+
+namespace MeterVault.Integration.Tests.Reconciliation;
+
+///
+/// Reconciles the water meter against the Wasser sheet's Wasserverbrauch column, exercising the
+/// mid-series register swap …861 → 2 (SDD §2.3). The importer detects the register drop and seeds
+/// a swap event whose consumption override comes from the sheet's own column.
+///
+public sealed class WaterReconciliationTests
+{
+ [Fact]
+ public void Water_consumption_matches_the_sheet_across_the_swap()
+ {
+ var staged = Stage(ReferenceProfiles.Water(), Water);
+
+ // A swap event must have been auto-detected at the 861→2 drop.
+ Assert.Contains(staged.Events, e => e.EventType == MeterEventType.MeterSwap);
+
+ var config = new MeterConfig
+ {
+ MeterId = ReferenceProfiles.Wasser,
+ Mode = MeterMode.CumulativeCounter,
+ Unit = "m3",
+ InitialBaseline = 820, // pre-existing meter's register when tracking began (Nov 2022).
+ };
+
+ var computed = ByMonth(Normalize(staged, config));
+ var oracle = OracleByMonth(ReadRows(Water), dateColumn: 0, valueColumn: 2, firstDataRow: 1);
+
+ AssertReconciles(computed, oracle, tolerance: 1.0, "water", minMatches: 12);
+
+ // Explicitly assert the swap month reconciles to 12.
+ Assert.Equal(12d, computed[new DateOnly(2023, 3, 1)], 3);
+ }
+}
diff --git a/tests/Integration.Tests/TimescaleFixture.cs b/tests/Integration.Tests/TimescaleFixture.cs
index e4e6207..53da599 100644
--- a/tests/Integration.Tests/TimescaleFixture.cs
+++ b/tests/Integration.Tests/TimescaleFixture.cs
@@ -35,6 +35,14 @@ public sealed class TimescaleFixture : IAsyncLifetime
await _db.StartAsync();
await using var ctx = CreateContext();
await ctx.Database.MigrateAsync();
+
+ // The fixtures are historical (2022–2023), so freshly-inserted chunks immediately fall
+ // past the 30-day compression horizon. Pause the compression job's scheduler so its
+ // background worker can't deadlock a test's import transaction. The policy still exists
+ // (SchemaTests verifies that); only its scheduling is disabled.
+ await ctx.Database.ExecuteSqlRawAsync(
+ "SELECT alter_job(job_id, scheduled => false) FROM timescaledb_information.jobs " +
+ "WHERE proc_name IN ('policy_compression', 'policy_columnstore');");
}
public async Task DisposeAsync() => await _db.DisposeAsync();