M4: aggregation + tariff-aware cost engine
- 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
This commit is contained in:
@@ -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)));
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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).
|
||||
/// </summary>
|
||||
[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<int> 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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user