From 6fc710d55a2b468af31a7d3723855bc094472dbc Mon Sep 17 00:00:00 2001 From: Florian Schmidt Date: Mon, 13 Jul 2026 11:52:53 +0200 Subject: [PATCH] M4: aggregation + tariff-aware cost engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ContinuousAggregates migration: consumption_daily/monthly/yearly in Europe/Berlin buckets via migrationBuilder.Sql(..., suppressTransaction: true), with refresh policies (end_offset >= 1 bucket; current bucket covered by real-time aggregation). - TariffResolver (Core): time-ranged price resolution, scope precedence meter > type > global. - CostService: Dapper-aggregated monthly consumption × resolved unit price (+base, -feed-in), month-dominant pricing; category rollups over member meters + meterless manual costs. - Tests: water cost reconciles to the sheet's Kosten column, Wasser category rollup (Dez=70€), monthly CAgg refresh matches base, TariffResolver unit tests. 91 tests green (56 Core + 35 integration). Claude-Session: https://claude.ai/code/session_01WujdMtMJPbxDpDnMeK22rr --- src/Core/Costing/TariffResolver.cs | 59 ++ src/Infrastructure/Costing/CostModels.cs | 15 + src/Infrastructure/Costing/CostService.cs | 134 +++ src/Infrastructure/DependencyInjection.cs | 1 + ...713094634_ContinuousAggregates.Designer.cs | 877 ++++++++++++++++++ .../20260713094634_ContinuousAggregates.cs | 61 ++ tests/Core.Tests/TariffResolverTests.cs | 57 ++ .../Costing/CostReconciliationTests.cs | 159 ++++ 8 files changed, 1363 insertions(+) create mode 100644 src/Core/Costing/TariffResolver.cs create mode 100644 src/Infrastructure/Costing/CostModels.cs create mode 100644 src/Infrastructure/Costing/CostService.cs create mode 100644 src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.Designer.cs create mode 100644 src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs create mode 100644 tests/Core.Tests/TariffResolverTests.cs create mode 100644 tests/Integration.Tests/Costing/CostReconciliationTests.cs diff --git a/src/Core/Costing/TariffResolver.cs b/src/Core/Costing/TariffResolver.cs new file mode 100644 index 0000000..06a3fd0 --- /dev/null +++ b/src/Core/Costing/TariffResolver.cs @@ -0,0 +1,59 @@ +using MeterVault.Core.Domain; + +namespace MeterVault.Core.Costing; + +/// +/// Resolves the tariff component in effect for a meter on a given date (SDD §7.5, FR-9). Scope +/// precedence is meter > energy type > global; within the winning scope the most recent +/// applicable valid_from wins. This is the "monthly dominant price" building block — the +/// caller passes the representative date for the period (e.g. mid-month). +/// +public static class TariffResolver +{ + public static Tariff? Resolve( + IEnumerable tariffs, TariffComponent component, int meterId, int energyTypeId, DateOnly date) + { + ArgumentNullException.ThrowIfNull(tariffs); + + Tariff? best = null; + var bestRank = 0; + foreach (var tariff in tariffs) + { + if (tariff.Component != component || tariff.ValidFrom > date) + { + continue; + } + + if (tariff.ValidTo is { } validTo && validTo < date) + { + continue; + } + + var rank = ScopeRank(tariff, meterId, energyTypeId); + if (rank == 0) + { + continue; + } + + if (rank > bestRank || (rank == bestRank && (best is null || tariff.ValidFrom > best.ValidFrom))) + { + best = tariff; + bestRank = rank; + } + } + + return best; + } + + public static double ResolveValue( + IEnumerable tariffs, TariffComponent component, int meterId, int energyTypeId, DateOnly date) => + Resolve(tariffs, component, meterId, energyTypeId, date)?.Value ?? 0d; + + private static int ScopeRank(Tariff tariff, int meterId, int energyTypeId) => tariff.ScopeType switch + { + TariffScope.Meter when tariff.ScopeId == meterId => 3, + TariffScope.EnergyType when tariff.ScopeId == energyTypeId => 2, + TariffScope.Global => 1, + _ => 0, + }; +} diff --git a/src/Infrastructure/Costing/CostModels.cs b/src/Infrastructure/Costing/CostModels.cs new file mode 100644 index 0000000..047605e --- /dev/null +++ b/src/Infrastructure/Costing/CostModels.cs @@ -0,0 +1,15 @@ +namespace MeterVault.Infrastructure.Costing; + +/// Cost and consumption/generation for one meter in one time bucket. +public sealed record MeterCostBucket(DateOnly Period, double Consumption, double Generation, double Cost); + +/// Rolled-up cost for a category in one time bucket (meters + manual costs). +public sealed record CategoryCostBucket(DateOnly Period, double Cost); + +/// Bucket granularity for cost/consumption queries. +public enum CostBucket +{ + Day, + Month, + Year, +} diff --git a/src/Infrastructure/Costing/CostService.cs b/src/Infrastructure/Costing/CostService.cs new file mode 100644 index 0000000..80f29cc --- /dev/null +++ b/src/Infrastructure/Costing/CostService.cs @@ -0,0 +1,134 @@ +using Dapper; +using MeterVault.Core.Costing; +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; + +namespace MeterVault.Infrastructure.Costing; + +/// +/// Computes cost by joining bucketed consumption with time-ranged tariffs (SDD §7.5). Consumption +/// is aggregated in SQL (Dapper, local-timezone buckets); the active price for each bucket is +/// resolved in C# via using the month's dominant price (SDD §14.3). +/// Categories roll up their member meters' costs plus meterless manual costs. +/// +public sealed class CostService(MeterVaultDbContext db) +{ + private readonly MeterVaultDbContext _db = db; + + public async Task> GetMeterCostsAsync( + int meterId, DateTimeOffset from, DateTimeOffset to, CostBucket bucket = CostBucket.Month, + CancellationToken cancellationToken = default) + { + var meter = await _db.Meters.FirstOrDefaultAsync(m => m.Id == meterId, cancellationToken).ConfigureAwait(false); + if (meter is null) + { + return []; + } + + var tariffs = await _db.Tariffs.AsNoTracking().ToListAsync(cancellationToken).ConfigureAwait(false); + var series = await QueryConsumptionAsync(meterId, from, to, bucket, cancellationToken).ConfigureAwait(false); + + var results = new List(); + foreach (var period in series.Keys.OrderBy(k => k)) + { + var (consumption, generation) = series[period]; + var mid = RepresentativeDate(period, bucket); + + var unitPrice = TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, meterId, meter.EnergyTypeId, mid); + var basePrice = TariffResolver.ResolveValue(tariffs, TariffComponent.BasePrice, meterId, meter.EnergyTypeId, mid); + var feedIn = TariffResolver.ResolveValue(tariffs, TariffComponent.FeedIn, meterId, meter.EnergyTypeId, mid); + + var cost = (consumption * unitPrice) + basePrice - (generation * feedIn); + results.Add(new MeterCostBucket(period, consumption, generation, cost)); + } + + return results; + } + + public async Task> GetCategoryCostsAsync( + int categoryId, DateTimeOffset from, DateTimeOffset to, CancellationToken cancellationToken = default) + { + var members = await _db.CostCategoryMembers + .Where(m => m.CategoryId == categoryId) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + var meterIds = new HashSet(); + foreach (var member in members) + { + if (member.MeterId is { } meterId) + { + meterIds.Add(meterId); + } + + if (member.EnergyTypeId is { } energyTypeId) + { + var byType = await _db.Meters.Where(m => m.EnergyTypeId == energyTypeId).Select(m => m.Id) + .ToListAsync(cancellationToken).ConfigureAwait(false); + meterIds.UnionWith(byType); + } + } + + var totals = new Dictionary(); + foreach (var meterId in meterIds) + { + foreach (var mc in await GetMeterCostsAsync(meterId, from, to, CostBucket.Month, cancellationToken).ConfigureAwait(false)) + { + totals[mc.Period] = totals.GetValueOrDefault(mc.Period) + mc.Cost; + } + } + + var manualCosts = await _db.ManualCosts + .Where(c => c.CategoryId == categoryId && c.PeriodStart >= DateOnly.FromDateTime(from.UtcDateTime) + && c.PeriodStart < DateOnly.FromDateTime(to.UtcDateTime)) + .ToListAsync(cancellationToken).ConfigureAwait(false); + foreach (var cost in manualCosts) + { + var period = new DateOnly(cost.PeriodStart.Year, cost.PeriodStart.Month, 1); + totals[period] = totals.GetValueOrDefault(period) + cost.Amount; + } + + return [.. totals.OrderBy(kv => kv.Key).Select(kv => new CategoryCostBucket(kv.Key, kv.Value))]; + } + + private async Task> QueryConsumptionAsync( + int meterId, DateTimeOffset from, DateTimeOffset to, CostBucket bucket, CancellationToken cancellationToken) + { + var interval = bucket switch + { + CostBucket.Day => "1 day", + CostBucket.Year => "1 year", + _ => "1 month", + }; + + var sql = + $"SELECT (time_bucket(INTERVAL '{interval}', \"time\", 'Europe/Berlin') AT TIME ZONE 'Europe/Berlin')::date AS period, " + + "kind, sum(amount) AS amount " + + "FROM consumption WHERE meter_id = @meterId AND \"time\" >= @from AND \"time\" < @to " + + "GROUP BY period, kind"; + + var connection = _db.Database.GetDbConnection(); + var command = new CommandDefinition(sql, new { meterId, from, to }, cancellationToken: cancellationToken); + var rows = await connection.QueryAsync(command).ConfigureAwait(false); + + var result = new Dictionary(); + foreach (var row in rows) + { + var current = result.GetValueOrDefault(row.Period); + result[row.Period] = row.Kind == (short)ConsumptionKind.Generation + ? (current.Item1, current.Item2 + row.Amount) + : (current.Item1 + row.Amount, current.Item2); + } + + return result; + } + + private static DateOnly RepresentativeDate(DateOnly period, CostBucket bucket) => bucket switch + { + CostBucket.Day => period, + CostBucket.Year => new DateOnly(period.Year, 7, 1), + _ => new DateOnly(period.Year, period.Month, 15), + }; + + private sealed record ConsumptionRow(DateOnly Period, short Kind, double Amount); +} diff --git a/src/Infrastructure/DependencyInjection.cs b/src/Infrastructure/DependencyInjection.cs index b68b7c4..491548c 100644 --- a/src/Infrastructure/DependencyInjection.cs +++ b/src/Infrastructure/DependencyInjection.cs @@ -27,6 +27,7 @@ public static class DependencyInjection services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddScoped(); return services; } diff --git a/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.Designer.cs b/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.Designer.cs new file mode 100644 index 0000000..ab1269d --- /dev/null +++ b/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.Designer.cs @@ -0,0 +1,877 @@ +// +using System; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace MeterVault.Infrastructure.Persistence.Migrations +{ + [DbContext(typeof(MeterVaultDbContext))] + [Migration("20260713094634_ContinuousAggregates")] + partial class ContinuousAggregates + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.HasPostgresExtension(modelBuilder, "timescaledb"); + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("MeterVault.Core.Domain.AppSetting", b => + { + b.Property("Key") + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("key"); + + b.Property("Value") + .IsRequired() + .HasColumnType("jsonb") + .HasColumnName("value"); + + b.HasKey("Key") + .HasName("pk_app_setting"); + + b.ToTable("app_setting", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Consumption", b => + { + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Time") + .HasColumnType("timestamp with time zone") + .HasColumnName("time"); + + b.Property("Kind") + .HasColumnType("smallint") + .HasColumnName("kind"); + + b.Property("Amount") + .HasColumnType("double precision") + .HasColumnName("amount"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("Quality") + .HasColumnType("smallint") + .HasColumnName("quality"); + + b.HasKey("MeterId", "Time", "Kind") + .HasName("pk_consumption"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_consumption_import_batch_id"); + + b.ToTable("consumption", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategory", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("ColorHex") + .HasColumnType("text") + .HasColumnName("color_hex"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("name"); + + b.Property("Sort") + .HasColumnType("integer") + .HasColumnName("sort"); + + b.HasKey("Id") + .HasName("pk_cost_category"); + + b.ToTable("cost_category", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategoryMember", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CategoryId") + .HasColumnType("integer") + .HasColumnName("category_id"); + + b.Property("EnergyTypeId") + .HasColumnType("smallint") + .HasColumnName("energy_type_id"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.HasKey("Id") + .HasName("pk_cost_category_member"); + + b.HasIndex("CategoryId") + .HasDatabaseName("ix_cost_category_member_category_id"); + + b.HasIndex("EnergyTypeId") + .HasDatabaseName("ix_cost_category_member_energy_type_id"); + + b.HasIndex("MeterId") + .HasDatabaseName("ix_cost_category_member_meter_id"); + + b.ToTable("cost_category_member", null, t => + { + t.HasCheckConstraint("ck_cost_category_member_target", "meter_id IS NOT NULL OR energy_type_id IS NOT NULL"); + }); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.EnergyType", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("smallint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("BaseUnit") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("base_unit"); + + b.Property("ColorHex") + .HasColumnType("text") + .HasColumnName("color_hex"); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("created_at") + .HasDefaultValueSql("now()"); + + b.Property("DefaultMode") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("default_mode"); + + b.Property("DisplayName") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("display_name"); + + b.Property("Icon") + .HasColumnType("text") + .HasColumnName("icon"); + + b.Property("Key") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("key"); + + b.HasKey("Id") + .HasName("pk_energy_type"); + + b.HasIndex("Key") + .IsUnique() + .HasDatabaseName("ix_energy_type_key"); + + b.ToTable("energy_type", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.ImportBatch", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("created_at") + .HasDefaultValueSql("now()"); + + b.Property("Mapping") + .HasColumnType("jsonb") + .HasColumnName("mapping"); + + b.Property("RevertedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("reverted_at"); + + b.Property("RowCount") + .HasColumnType("integer") + .HasColumnName("row_count"); + + b.Property("SourceName") + .HasColumnType("text") + .HasColumnName("source_name"); + + b.HasKey("Id") + .HasName("pk_import_batch"); + + b.ToTable("import_batch", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.IngestionEndpoint", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Config") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("config") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("IsEnabled") + .HasColumnType("boolean") + .HasColumnName("is_enabled"); + + b.Property("LastSeenAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_seen_at"); + + b.Property("LastStatus") + .HasColumnType("text") + .HasColumnName("last_status"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("name"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("type"); + + b.HasKey("Id") + .HasName("pk_ingestion_endpoint"); + + b.ToTable("ingestion_endpoint", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.ManualCost", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Amount") + .HasColumnType("double precision") + .HasColumnName("amount"); + + b.Property("CategoryId") + .HasColumnType("integer") + .HasColumnName("category_id"); + + b.Property("Currency") + .IsRequired() + .HasMaxLength(8) + .HasColumnType("character varying(8)") + .HasColumnName("currency"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Notes") + .HasColumnType("text") + .HasColumnName("notes"); + + b.Property("PeriodEnd") + .HasColumnType("date") + .HasColumnName("period_end"); + + b.Property("PeriodStart") + .HasColumnType("date") + .HasColumnName("period_start"); + + b.HasKey("Id") + .HasName("pk_manual_cost"); + + b.HasIndex("CategoryId") + .HasDatabaseName("ix_manual_cost_category_id"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_manual_cost_import_batch_id"); + + b.HasIndex("MeterId") + .HasDatabaseName("ix_manual_cost_meter_id"); + + b.ToTable("manual_cost", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Meter", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("created_at") + .HasDefaultValueSql("now()"); + + b.Property("EnergyTypeId") + .HasColumnType("smallint") + .HasColumnName("energy_type_id"); + + b.Property("InitialBaseline") + .HasColumnType("double precision") + .HasColumnName("initial_baseline"); + + b.Property("InstalledAt") + .HasColumnType("date") + .HasColumnName("installed_at"); + + b.Property("IsActive") + .ValueGeneratedOnAdd() + .HasColumnType("boolean") + .HasDefaultValue(true) + .HasColumnName("is_active"); + + b.Property("Location") + .HasColumnType("text") + .HasColumnName("location"); + + b.Property("Manufacturer") + .HasColumnType("text") + .HasColumnName("manufacturer"); + + b.Property("Meta") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("meta") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("Mode") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("mode"); + + b.Property("Model") + .HasColumnType("text") + .HasColumnName("model"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text") + .HasColumnName("name"); + + b.Property("RetiredAt") + .HasColumnType("date") + .HasColumnName("retired_at"); + + b.Property("SerialNumber") + .HasColumnType("text") + .HasColumnName("serial_number"); + + b.Property("Unit") + .IsRequired() + .HasColumnType("text") + .HasColumnName("unit"); + + b.Property("UpdatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("updated_at") + .HasDefaultValueSql("now()"); + + b.HasKey("Id") + .HasName("pk_meter"); + + b.HasIndex("EnergyTypeId", "IsActive") + .HasDatabaseName("ix_meter_energy_type_id_is_active"); + + b.ToTable("meter", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterEvent", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Amount") + .HasColumnType("double precision") + .HasColumnName("amount"); + + b.Property("EventType") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("event_type"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("Meta") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("meta") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("NewValue") + .HasColumnType("double precision") + .HasColumnName("new_value"); + + b.Property("Notes") + .HasColumnType("text") + .HasColumnName("notes"); + + b.Property("PrevValue") + .HasColumnType("double precision") + .HasColumnName("prev_value"); + + b.Property("Time") + .HasColumnType("timestamp with time zone") + .HasColumnName("time"); + + b.Property("Unit") + .HasColumnType("text") + .HasColumnName("unit"); + + b.HasKey("Id") + .HasName("pk_meter_event"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_meter_event_import_batch_id"); + + b.HasIndex("MeterId", "Time") + .HasDatabaseName("ix_meter_event_meter_id_time"); + + b.ToTable("meter_event", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterSource", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Config") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("config") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("EndpointId") + .HasColumnType("integer") + .HasColumnName("endpoint_id"); + + b.Property("IsEnabled") + .HasColumnType("boolean") + .HasColumnName("is_enabled"); + + b.Property("LastSeenAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_seen_at"); + + b.Property("LastStatus") + .HasColumnType("text") + .HasColumnName("last_status"); + + b.Property("LastValue") + .HasColumnType("double precision") + .HasColumnName("last_value"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Offset") + .HasColumnType("double precision") + .HasColumnName("offset"); + + b.Property("Priority") + .HasColumnType("integer") + .HasColumnName("priority"); + + b.Property("Scale") + .ValueGeneratedOnAdd() + .HasColumnType("double precision") + .HasDefaultValue(1.0) + .HasColumnName("scale"); + + b.Property("SourceType") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("source_type"); + + b.Property("ValueKind") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("value_kind"); + + b.HasKey("Id") + .HasName("pk_meter_source"); + + b.HasIndex("EndpointId") + .HasDatabaseName("ix_meter_source_endpoint_id"); + + b.HasIndex("MeterId") + .HasDatabaseName("ix_meter_source_meter_id"); + + b.ToTable("meter_source", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Reading", b => + { + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Time") + .HasColumnType("timestamp with time zone") + .HasColumnName("time"); + + b.Property("Flags") + .HasColumnType("integer") + .HasColumnName("flags"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("Quality") + .HasColumnType("smallint") + .HasColumnName("quality"); + + b.Property("SourceId") + .HasColumnType("integer") + .HasColumnName("source_id"); + + b.Property("Value") + .HasColumnType("double precision") + .HasColumnName("value"); + + b.HasKey("MeterId", "Time") + .HasName("pk_reading"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_reading_import_batch_id"); + + b.ToTable("reading", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Tank", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CachedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("cached_at"); + + b.Property("CachedBalance") + .HasColumnType("double precision") + .HasColumnName("cached_balance"); + + b.Property("Calibration") + .HasColumnType("jsonb") + .HasColumnName("calibration"); + + b.Property("Capacity") + .HasColumnType("double precision") + .HasColumnName("capacity"); + + b.Property("FixedRate") + .HasColumnType("double precision") + .HasColumnName("fixed_rate"); + + b.Property("LowThreshold") + .HasColumnType("double precision") + .HasColumnName("low_threshold"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("RateMode") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("rate_mode"); + + b.Property("ReorderThreshold") + .HasColumnType("double precision") + .HasColumnName("reorder_threshold"); + + b.Property("Unit") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("unit"); + + b.HasKey("Id") + .HasName("pk_tank"); + + b.HasIndex("MeterId") + .IsUnique() + .HasDatabaseName("ix_tank_meter_id"); + + b.ToTable("tank", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Tariff", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Component") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("component"); + + b.Property("Currency") + .IsRequired() + .HasMaxLength(8) + .HasColumnType("character varying(8)") + .HasColumnName("currency"); + + b.Property("Notes") + .HasColumnType("text") + .HasColumnName("notes"); + + b.Property("ScopeId") + .HasColumnType("integer") + .HasColumnName("scope_id"); + + b.Property("ScopeType") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("scope_type"); + + b.Property("Unit") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("unit"); + + b.Property("ValidFrom") + .HasColumnType("date") + .HasColumnName("valid_from"); + + b.Property("ValidTo") + .HasColumnType("date") + .HasColumnName("valid_to"); + + b.Property("Value") + .HasColumnType("double precision") + .HasColumnName("value"); + + b.HasKey("Id") + .HasName("pk_tariff"); + + b.HasIndex("ScopeType", "ScopeId", "Component", "ValidFrom") + .HasDatabaseName("ix_tariff_scope_type_scope_id_component_valid_from"); + + b.ToTable("tariff", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Consumption", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired() + .HasConstraintName("fk_consumption_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategoryMember", b => + { + b.HasOne("MeterVault.Core.Domain.CostCategory", "Category") + .WithMany("Members") + .HasForeignKey("CategoryId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_cost_category_member_cost_category_category_id"); + + b.HasOne("MeterVault.Core.Domain.EnergyType", null) + .WithMany() + .HasForeignKey("EnergyTypeId") + .OnDelete(DeleteBehavior.Cascade) + .HasConstraintName("fk_cost_category_member_energy_type_energy_type_id"); + + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .HasConstraintName("fk_cost_category_member_meter_meter_id"); + + b.Navigation("Category"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.ManualCost", b => + { + b.HasOne("MeterVault.Core.Domain.CostCategory", null) + .WithMany() + .HasForeignKey("CategoryId") + .OnDelete(DeleteBehavior.SetNull) + .HasConstraintName("fk_manual_cost_cost_category_category_id"); + + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.SetNull) + .HasConstraintName("fk_manual_cost_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Meter", b => + { + b.HasOne("MeterVault.Core.Domain.EnergyType", "EnergyType") + .WithMany("Meters") + .HasForeignKey("EnergyTypeId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired() + .HasConstraintName("fk_meter_energy_type_energy_type_id"); + + b.Navigation("EnergyType"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterEvent", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_meter_event_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterSource", b => + { + b.HasOne("MeterVault.Core.Domain.IngestionEndpoint", "Endpoint") + .WithMany() + .HasForeignKey("EndpointId") + .OnDelete(DeleteBehavior.SetNull) + .HasConstraintName("fk_meter_source_ingestion_endpoints_endpoint_id"); + + b.HasOne("MeterVault.Core.Domain.Meter", "Meter") + .WithMany("Sources") + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_meter_source_meter_meter_id"); + + b.Navigation("Endpoint"); + + b.Navigation("Meter"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Reading", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired() + .HasConstraintName("fk_reading_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Tank", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", "Meter") + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_tank_meter_meter_id"); + + b.Navigation("Meter"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategory", b => + { + b.Navigation("Members"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.EnergyType", b => + { + b.Navigation("Meters"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Meter", b => + { + b.Navigation("Sources"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs b/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs new file mode 100644 index 0000000..99df849 --- /dev/null +++ b/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs @@ -0,0 +1,61 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace MeterVault.Infrastructure.Persistence.Migrations +{ + /// + /// Timescale continuous aggregates that roll normalized consumption up to daily / monthly / + /// yearly buckets in the instance timezone (SDD §5.4, §10). Dashboards read these, never the + /// raw hypertable. CREATE MATERIALIZED VIEW … WITH (timescaledb.continuous) and the policy + /// calls cannot run inside a transaction, so each is issued with suppressTransaction: true and + /// as its own statement; once-only execution is guaranteed by __EFMigrationsHistory. + /// + public partial class ContinuousAggregates : Migration + { + private const string Timezone = "Europe/Berlin"; + + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + CreateAggregate(migrationBuilder, "consumption_daily", "1 day"); + CreateAggregate(migrationBuilder, "consumption_monthly", "1 month"); + CreateAggregate(migrationBuilder, "consumption_yearly", "1 year"); + + // end_offset must be at least one bucket wide; the current (incomplete) bucket is still + // reflected at query time via Timescale real-time aggregation. + AddPolicy(migrationBuilder, "consumption_daily", startOffset: "30 days", endOffset: "1 day"); + AddPolicy(migrationBuilder, "consumption_monthly", startOffset: "1 year", endOffset: "1 month"); + AddPolicy(migrationBuilder, "consumption_yearly", startOffset: "10 years", endOffset: "1 year"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + // Dropping the materialized view removes its refresh policy job too. + migrationBuilder.Sql("DROP MATERIALIZED VIEW IF EXISTS consumption_yearly;", suppressTransaction: true); + migrationBuilder.Sql("DROP MATERIALIZED VIEW IF EXISTS consumption_monthly;", suppressTransaction: true); + migrationBuilder.Sql("DROP MATERIALIZED VIEW IF EXISTS consumption_daily;", suppressTransaction: true); + } + + private static void CreateAggregate(MigrationBuilder builder, string name, string bucket) + { + builder.Sql( + $"CREATE MATERIALIZED VIEW {name} WITH (timescaledb.continuous) AS " + + $"SELECT time_bucket(INTERVAL '{bucket}', time, '{Timezone}') AS bucket, " + + "meter_id, kind, sum(amount) AS amount " + + "FROM consumption GROUP BY bucket, meter_id, kind WITH NO DATA;", + suppressTransaction: true); + } + + private static void AddPolicy(MigrationBuilder builder, string name, string startOffset, string endOffset) + { + builder.Sql( + $"SELECT add_continuous_aggregate_policy('{name}', " + + $"start_offset => INTERVAL '{startOffset}', " + + $"end_offset => INTERVAL '{endOffset}', " + + "schedule_interval => INTERVAL '1 hour');", + suppressTransaction: true); + } + } +} diff --git a/tests/Core.Tests/TariffResolverTests.cs b/tests/Core.Tests/TariffResolverTests.cs new file mode 100644 index 0000000..2aaf15f --- /dev/null +++ b/tests/Core.Tests/TariffResolverTests.cs @@ -0,0 +1,57 @@ +using MeterVault.Core.Costing; +using MeterVault.Core.Domain; + +namespace MeterVault.Core.Tests; + +public sealed class TariffResolverTests +{ + private static Tariff Unit(double value, TariffScope scope, int? scopeId, DateOnly from, DateOnly? to = null) => new() + { + Component = TariffComponent.UnitPrice, + Value = value, + Unit = "EUR/kWh", + ScopeType = scope, + ScopeId = scopeId, + ValidFrom = from, + ValidTo = to, + }; + + [Fact] + public void Picks_the_price_valid_for_the_date() + { + // Electricity price history from the Strom sheet. + var tariffs = new[] + { + Unit(0.16, TariffScope.EnergyType, 1, new DateOnly(2022, 9, 1), new DateOnly(2022, 12, 31)), + Unit(0.44, TariffScope.EnergyType, 1, new DateOnly(2023, 1, 1), new DateOnly(2023, 4, 30)), + Unit(0.37, TariffScope.EnergyType, 1, new DateOnly(2023, 5, 1)), + }; + + Assert.Equal(0.16, TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, 5, 1, new DateOnly(2022, 10, 15))); + Assert.Equal(0.44, TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, 5, 1, new DateOnly(2023, 2, 15))); + Assert.Equal(0.37, TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, 5, 1, new DateOnly(2024, 6, 15))); + } + + [Fact] + public void Meter_scope_overrides_energy_type_and_global() + { + var tariffs = new[] + { + Unit(0.30, TariffScope.Global, null, new DateOnly(2023, 1, 1)), + Unit(0.40, TariffScope.EnergyType, 1, new DateOnly(2023, 1, 1)), + Unit(0.50, TariffScope.Meter, 7, new DateOnly(2023, 1, 1)), + }; + + Assert.Equal(0.50, TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, 7, 1, new DateOnly(2023, 6, 1))); + Assert.Equal(0.40, TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, 8, 1, new DateOnly(2023, 6, 1))); + Assert.Equal(0.30, TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, 8, 2, new DateOnly(2023, 6, 1))); + } + + [Fact] + public void Returns_zero_when_nothing_applies() + { + var tariffs = new[] { Unit(0.25, TariffScope.EnergyType, 1, new DateOnly(2024, 1, 1)) }; + + Assert.Equal(0d, TariffResolver.ResolveValue(tariffs, TariffComponent.UnitPrice, 1, 1, new DateOnly(2023, 1, 1))); + } +} diff --git a/tests/Integration.Tests/Costing/CostReconciliationTests.cs b/tests/Integration.Tests/Costing/CostReconciliationTests.cs new file mode 100644 index 0000000..70c20e1 --- /dev/null +++ b/tests/Integration.Tests/Costing/CostReconciliationTests.cs @@ -0,0 +1,159 @@ +using MeterVault.Core.Domain; +using MeterVault.Core.Normalization; +using MeterVault.Infrastructure.Costing; +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.Costing; + +/// +/// Reconciles the cost engine against the Wasser sheet's Kosten column (consumption × €/m³) and +/// checks category rollups and continuous-aggregate refresh (SDD §7.5, §5.4). +/// +[Collection("Timescale")] +public sealed class CostReconciliationTests(TimescaleFixture fx) +{ + private static readonly DateTimeOffset From = new(2022, 11, 1, 0, 0, 0, TimeSpan.Zero); + private static readonly DateTimeOffset To = new(2024, 1, 1, 0, 0, 0, TimeSpan.Zero); + + [Fact] + public async Task Water_cost_matches_the_sheet() + { + await using var db = fx.CreateContext(); + var meterId = await ImportWaterAsync(db); + + db.Tariffs.Add(new Tariff + { + ScopeType = TariffScope.Meter, + ScopeId = meterId, + Component = TariffComponent.UnitPrice, + Value = 5.00, + Unit = "EUR/m3", + ValidFrom = new DateOnly(2022, 11, 1), + }); + await db.SaveChangesAsync(); + + var costs = await new CostService(db).GetMeterCostsAsync(meterId, From, To); + var computed = costs.ToDictionary(c => c.Period, c => c.Cost); + var oracle = OracleByMonth(ReadRows(Water), dateColumn: 0, valueColumn: 4, firstDataRow: 1); // Kosten + + AssertReconciles(computed, oracle, tolerance: 0.02, "water cost", minMatches: 10); + + await CleanupAsync(db, meterId); + } + + [Fact] + public async Task Category_rollup_includes_meter_costs() + { + await using var db = fx.CreateContext(); + var meterId = await ImportWaterAsync(db); + var wasserCategory = await db.CostCategories.FirstAsync(c => c.Name == "Wasser"); + + db.Tariffs.Add(new Tariff + { + ScopeType = TariffScope.Meter, + ScopeId = meterId, + Component = TariffComponent.UnitPrice, + Value = 5.00, + Unit = "EUR/m3", + ValidFrom = new DateOnly(2022, 11, 1), + }); + db.CostCategoryMembers.Add(new CostCategoryMember { CategoryId = wasserCategory.Id, MeterId = meterId }); + await db.SaveChangesAsync(); + + var rollup = await new CostService(db).GetCategoryCostsAsync(wasserCategory.Id, From, To); + var dec2022 = rollup.Single(r => r.Period == new DateOnly(2022, 12, 1)); + + Assert.Equal(70d, dec2022.Cost, 2); // Dez 2022: 14 m³ × 5,00 € + + await db.CostCategoryMembers.Where(m => m.MeterId == meterId).ExecuteDeleteAsync(); + await CleanupAsync(db, meterId); + } + + [Fact] + public async Task Monthly_continuous_aggregate_refreshes_and_matches_base() + { + await using var db = fx.CreateContext(); + var meterId = await ImportWaterAsync(db); + + // refresh_continuous_aggregate cannot run inside a transaction — use the raw connection. + var connection = db.Database.GetDbConnection(); + await connection.OpenAsync(); + await using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = "CALL refresh_continuous_aggregate('consumption_monthly', NULL, NULL);"; + await cmd.ExecuteNonQueryAsync(); + } + + double aggregated; + await using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = + "SELECT sum(amount) FROM consumption_monthly WHERE meter_id = @m " + + "AND (bucket AT TIME ZONE 'Europe/Berlin')::date = DATE '2022-12-01';"; + var p = cmd.CreateParameter(); + p.ParameterName = "m"; + p.Value = meterId; + cmd.Parameters.Add(p); + aggregated = Convert.ToDouble(await cmd.ExecuteScalarAsync()); + } + + Assert.Equal(14d, aggregated, 1); // Dez 2022 consumption + + await CleanupAsync(db, meterId); + } + + private static async Task ImportWaterAsync(MeterVaultDbContext db) + { + await DatabaseSeeder.SeedAsync(db); + var waterType = await db.EnergyTypes.FirstAsync(t => t.Key == "water"); + var meter = new Meter + { + Name = $"cost-water-{Guid.NewGuid():N}", + EnergyTypeId = waterType.Id, + Mode = MeterMode.CumulativeCounter, + Unit = "m3", + InitialBaseline = 820, + }; + db.Meters.Add(meter); + await db.SaveChangesAsync(); + + var profile = new MappingProfile + { + Name = "cost-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 service = new ImportService(db, new NormalizationService(db, NormalizationEngine.CreateDefault())); + await service.CommitAsync(staged, "Wasser.csv", null); + return meter.Id; + } + + private static async Task CleanupAsync(MeterVaultDbContext db, int meterId) + { + await db.Consumption.Where(c => c.MeterId == meterId).ExecuteDeleteAsync(); + await db.Readings.Where(r => r.MeterId == meterId).ExecuteDeleteAsync(); + await db.MeterEvents.Where(e => e.MeterId == meterId).ExecuteDeleteAsync(); + await db.Tariffs.Where(t => t.ScopeId == meterId && t.ScopeType == TariffScope.Meter).ExecuteDeleteAsync(); + await db.Meters.Where(m => m.Id == meterId).ExecuteDeleteAsync(); + } +}