FractalDataWorks.Calculations 0.6.0-rc.1

FractalDataWorks.Calculations

Core calculation implementations for the FractalDataWorks calculation framework. This package provides the concrete implementation of ICalculationContext and supporting infrastructure for executing calculations with data access capabilities.

Overview

FractalDataWorks.Calculations builds on the abstractions defined in FractalDataWorks.Calculations.Abstractions to provide:

• CalculationContext: Default implementation of ICalculationContext with IDataGateway integration
• Execution Infrastructure: Support for parameter passing and state management
• Data Access Integration: Seamless integration with FractalDataWorks data services
• Railway-Oriented Execution: Built-in support for IGenericResult patterns

This package serves as the runtime for executing calculations defined using the calculation abstractions.

Target Frameworks: .NET Standard 2.0 Dependencies:

• FractalDataWorks.Calculations.Abstractions
• FractalDataWorks.Services.Data.Abstractions
• FractalDataWorks.Commands.Data.Abstractions
• FractalDataWorks.Results

Core Components

CalculationContext

The default implementation of ICalculationContext providing execution context for calculations.

From CalculationContext.cs:16-50:

[ExcludeFromCodeCoverage]
public sealed class CalculationContext : ICalculationContext
{
    private readonly Dictionary<string, object?> _state;

    public CalculationContext(
        IDataGateway dataGateway,
        IReadOnlyDictionary<string, object?>? parameters = null)
    {
        DataGateway = dataGateway ?? throw new ArgumentNullException(nameof(dataGateway));
        ExecutionId = Guid.NewGuid().ToString();
        ExecutionTime = DateTime.UtcNow;
        Parameters = parameters ?? new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
        _state = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
    }

    public string ExecutionId { get; }
    public DateTime ExecutionTime { get; }
    public IDataGateway DataGateway { get; }
    public IReadOnlyDictionary<string, object?> Parameters { get; }
    public IDictionary<string, object?> State => _state;
}

Key Features:

• Automatic execution ID generation for tracking
• UTC timestamp for execution time
• Case-insensitive parameter and state dictionaries
• Direct access to IDataGateway for data operations
• Supports both connection-based and command-based data retrieval

Dataset Convenience Methods

The context provides typed methods for caching datasets during calculation execution.

From CalculationContext.cs:82-122:

// Get a cached dataset by name
T? GetDataSet<T>(string name);

// Try to get a cached dataset (returns false if not found)
bool TryGetDataSet<T>(string name, out T? dataset);

// Store a dataset in the cache
void SetDataSet<T>(string name, T dataset);

// Check if a dataset exists in the cache
bool HasDataSet(string name);

Calculation Result Convenience Methods

Store and retrieve results from dependent calculations.

From CalculationContext.cs:125-165:

// Get a cached calculation result
T? GetCalculationResult<T>(string calculationName);

// Try to get a cached calculation result
bool TryGetCalculationResult<T>(string calculationName, out T? result);

// Store a calculation result
void SetCalculationResult<T>(string calculationName, T result);

// Check if a calculation result exists
bool HasCalculationResult(string calculationName);

Data Access Methods

Execute data commands through the context.

From CalculationContext.cs:53-80:

// Execute a data command through the DataGateway
public async Task<IGenericResult<TData>> GetData<TData>(
    IDataCommand command,
    CancellationToken cancellationToken = default)
{
    if (command == null)
        return GenericResult<TData>.Failure("Command is required");

    return await DataGateway.Execute<TData>(command, cancellationToken).ConfigureAwait(false);
}

Usage Patterns

The following examples demonstrate common patterns for using CalculationContext. These patterns are based on the actual API defined in ICalculationContext.cs and ICalculation.cs.

Basic Context Creation

// Inject IDataGateway from DI
public class CalculationService
{
    private readonly IDataGateway _dataGateway;

    public CalculationService(IDataGateway dataGateway)
    {
        _dataGateway = dataGateway;
    }

    public async Task<IGenericResult<decimal>> ExecuteCalculation(
        ICalculation<IEnumerable<decimal>, decimal> calculation,
        IEnumerable<decimal> input)
    {
        // Create context
        var context = new CalculationContext(_dataGateway);

        // Execute calculation
        var result = await calculation.Execute(input, context, CancellationToken.None);

        return result;
    }
}

Context with Parameters

// Pass configuration parameters to calculation
var parameters = new Dictionary<string, object?>
{
    ["Year"] = 2024,
    ["Region"] = "North",
    ["IncludeTax"] = true,
    ["Currency"] = "USD"
};

var context = new CalculationContext(_dataGateway, parameters);

// Calculation can access parameters
public async Task<IGenericResult<decimal>> Execute(
    string customerId,
    ICalculationContext context,
    CancellationToken cancellationToken)
{
    var year = (int)context.Parameters["Year"];
    var region = (string)context.Parameters["Region"];

    // Use parameters in data query
    var command = new QueryCommand<SalesData>($"Sales_{year}")
    {
        Filter = $"CustomerId = '{customerId}' AND Region = '{region}'"
    };

    var data = await context.GetData<IEnumerable<SalesData>>(command, cancellationToken);
    // ... calculation logic
}

Using Convenience Methods for Calculation Chains

Use the typed convenience methods for storing and retrieving datasets and calculation results:

public async Task<IGenericResult<decimal>> Execute(
    string productId,
    ICalculationContext context,
    CancellationToken cancellationToken)
{
    // Check if sales data is already cached
    if (!context.TryGetDataSet<IEnumerable<SalesRecord>>("SalesData", out var salesData))
    {
        // Load and cache the dataset
        var command = new QueryCommand<SalesRecord>("Sales") { ConnectionName = "Primary" };
        var result = await context.GetData<IEnumerable<SalesRecord>>(command, cancellationToken);
        if (!result.IsSuccess)
            return GenericResult<decimal>.Failure(result.CurrentMessage);

        salesData = result.Value;
        context.SetDataSet("SalesData", salesData);
    }

    // Check if a dependent calculation result exists
    if (context.TryGetCalculationResult<decimal>("TotalCosts", out var costs))
    {
        // Use cached result from previous calculation
        var totalSales = salesData.Sum(s => s.Amount);
        var profit = totalSales - costs;

        // Store this calculation's result for downstream calculations
        context.SetCalculationResult("Profit", profit);

        return GenericResult<decimal>.Success(profit);
    }

    return GenericResult<decimal>.Failure("Dependent calculation 'TotalCosts' not found");
}

Using State for Intermediate Values

For ad-hoc state storage during calculation execution:

public async Task<IGenericResult<CalculationResult>> Execute(
    string productId,
    ICalculationContext context,
    CancellationToken cancellationToken)
{
    // Store timing information
    context.State["StartTime"] = DateTime.UtcNow;

    // Execute calculation logic...
    var result = await PerformCalculation(productId, context, cancellationToken);

    context.State["EndTime"] = DateTime.UtcNow;

    return result;
}

Data Access Patterns

Using Command-Based Data Access

public async Task<IGenericResult<decimal>> Execute(
    string customerId,
    ICalculationContext context,
    CancellationToken cancellationToken)
{
    // Create specific data command
    var command = new QueryCommand<Order>("Orders")
    {
        ConnectionName = "PrimaryDb",
        Filter = $"CustomerId = '{customerId}' AND Status = 'Completed'",
        OrderBy = "OrderDate DESC"
    };

    // Execute through context
    var ordersResult = await context.GetData<IEnumerable<Order>>(command, cancellationToken);

    if (!ordersResult.IsSuccess)
        return GenericResult<decimal>.Failure($"Failed to retrieve orders: {ordersResult.CurrentMessage}");

    var totalRevenue = ordersResult.Value.Sum(o => o.Total);
    return GenericResult<decimal>.Success(totalRevenue);
}

Using Connection-Based Data Access

// Note: Basic connection-based access returns failure by default
// Use command-based approach instead
var result = await context.GetData<OrderData>("PrimaryDb", "Orders", cancellationToken);

// This returns:
// GenericResult<OrderData>.Failure("Use GetData<TData>(IDataCommand) overload for query execution")

Multi-Source Data Aggregation

public async Task<IGenericResult<CombinedMetrics>> Execute(
    string entityId,
    ICalculationContext context,
    CancellationToken cancellationToken)
{
    // Get data from SQL database
    var sqlCommand = new QueryCommand<SalesRecord>("Sales")
    {
        ConnectionName = "SqlServer",
        Filter = $"EntityId = '{entityId}'"
    };
    var salesData = await context.GetData<IEnumerable<SalesRecord>>(sqlCommand, cancellationToken);

    // Get data from REST API
    var apiCommand = new RestQueryCommand<AnalyticsData>($"analytics/{entityId}")
    {
        ConnectionName = "AnalyticsApi"
    };
    var analyticsData = await context.GetData<AnalyticsData>(apiCommand, cancellationToken);

    // Get data from file storage
    var fileCommand = new FileReadCommand<CustomerData>($"customers/{entityId}.json")
    {
        ConnectionName = "FileStore"
    };
    var customerData = await context.GetData<CustomerData>(fileCommand, cancellationToken);

    // Combine results
    if (salesData.IsSuccess && analyticsData.IsSuccess && customerData.IsSuccess)
    {
        var metrics = new CombinedMetrics
        {
            TotalSales = salesData.Value.Sum(s => s.Amount),
            PageViews = analyticsData.Value.Views,
            CustomerLifetimeValue = customerData.Value.LifetimeValue
        };

        return GenericResult<CombinedMetrics>.Success(metrics);
    }

    return GenericResult<CombinedMetrics>.Failure("Failed to retrieve all required data");
}

Service Registration

public static class ServiceCollectionExtensions
{
    public static IServiceCollection AddCalculationServices(
        this IServiceCollection services)
    {
        // Register CalculationContext
        services.AddScoped<ICalculationContext>(sp =>
        {
            var dataGateway = sp.GetRequiredService<IDataGateway>();
            return new CalculationContext(dataGateway);
        });

        // Or with parameters from configuration
        services.AddScoped<ICalculationContext>(sp =>
        {
            var dataGateway = sp.GetRequiredService<IDataGateway>();
            var config = sp.GetRequiredService<IConfiguration>();

            var parameters = new Dictionary<string, object?>
            {
                ["DefaultYear"] = config.GetValue<int>("Calculations:DefaultYear"),
                ["DefaultRegion"] = config.GetValue<string>("Calculations:DefaultRegion")
            };

            return new CalculationContext(dataGateway, parameters);
        });

        return services;
    }
}

Integration with ETL Pipelines

public class CalculationStage : ITransformStage
{
    private readonly ICalculationContext _context;
    private readonly ICalculation<IEnumerable<decimal>, decimal> _calculation;

    public CalculationStage(
        ICalculationContext context,
        ICalculation<IEnumerable<decimal>, decimal> calculation)
    {
        _context = context;
        _calculation = calculation;
    }

    public async Task<IGenericResult> Execute(
        IStageExecutionContext stageContext,
        CancellationToken cancellationToken)
    {
        // Get input from previous stage
        var inputData = stageContext.InputData["Metrics"] as IEnumerable<decimal>;

        // Execute calculation
        var result = await _calculation.Execute(inputData, _context, cancellationToken);

        if (!result.IsSuccess)
            return GenericResult.Failure($"Calculation failed: {result.CurrentMessage}");

        // Store output for next stage
        stageContext.OutputData["CalculatedValue"] = result.Value;

        return GenericResult.Success();
    }
}

Execution Tracking

public class TrackedCalculationExecutor
{
    private readonly ILogger<TrackedCalculationExecutor> _logger;
    private readonly IDataGateway _dataGateway;

    public async Task<IGenericResult<TOutput>> ExecuteWithTracking<TInput, TOutput>(
        ICalculation<TInput, TOutput> calculation,
        TInput input,
        IReadOnlyDictionary<string, object?>? parameters = null,
        CancellationToken cancellationToken = default)
    {
        var context = new CalculationContext(_dataGateway, parameters);

        _logger.LogInformation(
            "Executing calculation {CalculationId} ({CalculationName}) with ExecutionId {ExecutionId}",
            calculation.CalculationId,
            calculation.Name,
            context.ExecutionId);

        var startTime = DateTime.UtcNow;

        try
        {
            var result = await calculation.Execute(input, context, cancellationToken);

            var duration = DateTime.UtcNow - startTime;

            if (result.IsSuccess)
            {
                _logger.LogInformation(
                    "Calculation {ExecutionId} completed successfully in {Duration}ms",
                    context.ExecutionId,
                    duration.TotalMilliseconds);
            }
            else
            {
                _logger.LogWarning(
                    "Calculation {ExecutionId} failed: {Error}",
                    context.ExecutionId,
                    result.CurrentMessage);
            }

            return result;
        }
        catch (Exception ex)
        {
            var duration = DateTime.UtcNow - startTime;
            _logger.LogError(ex,
                "Calculation {ExecutionId} threw exception after {Duration}ms",
                context.ExecutionId,
                duration.TotalMilliseconds);

            return GenericResult<TOutput>.Failure($"Calculation threw exception: {ex.Message}");
        }
    }
}

Best Practices

1. Inject IDataGateway: Always inject IDataGateway through DI, never create instances directly
2. Reuse Context: Create one context per calculation execution, not per data access
3. Parameter Validation: Validate parameters exist before using them in calculations
4. Use Convenience Methods: Prefer GetDataSet<T>() and GetCalculationResult<T>() over direct State dictionary access for type safety
5. Execution IDs: Log the ExecutionId for tracking and correlation
6. Cancellation Support: Always pass CancellationToken to GetData calls
7. Result Checking: Always check IsSuccess before accessing result.Value
8. Connection Names: Use named connections from configuration, not hardcoded strings
9. Validate Dependencies: Call ValidateDependencies(context) before execution to verify required datasets and calculations are available

Error Handling

public async Task<IGenericResult<TOutput>> SafeExecute<TInput, TOutput>(
    ICalculation<TInput, TOutput> calculation,
    TInput input,
    IReadOnlyDictionary<string, object?>? parameters = null)
{
    try
    {
        var context = new CalculationContext(_dataGateway, parameters);

        // Validate calculation first
        var validationResult = await calculation.Validate(CancellationToken.None);
        if (!validationResult.IsSuccess)
        {
            return GenericResult<TOutput>.Failure(
                $"Validation failed: {validationResult.CurrentMessage}");
        }

        // Execute calculation
        return await calculation.Execute(input, context, CancellationToken.None);
    }
    catch (ArgumentNullException ex)
    {
        return GenericResult<TOutput>.Failure($"Invalid argument: {ex.ParamName}");
    }
    catch (InvalidOperationException ex)
    {
        return GenericResult<TOutput>.Failure($"Invalid operation: {ex.Message}");
    }
    catch (Exception ex)
    {
        return GenericResult<TOutput>.Failure($"Unexpected error: {ex.Message}");
    }
}

Testing

public class CalculationContextTests
{
    [Fact]
    public void ContextGeneratesUniqueExecutionId()
    {
        // Arrange
        var mockDataGateway = new Mock<IDataGateway>();

        // Act
        var context1 = new CalculationContext(mockDataGateway.Object);
        var context2 = new CalculationContext(mockDataGateway.Object);

        // Assert
        context1.ExecutionId.ShouldNotBe(context2.ExecutionId);
    }

    [Fact]
    public void SetDataSetStoresAndRetrievesTypedData()
    {
        // Arrange
        var mockDataGateway = new Mock<IDataGateway>();
        var context = new CalculationContext(mockDataGateway.Object);
        var testData = new List<int> { 1, 2, 3 };

        // Act
        context.SetDataSet("TestData", testData);

        // Assert
        context.HasDataSet("TestData").ShouldBeTrue();
        context.GetDataSet<List<int>>("TestData").ShouldBe(testData);
    }

    [Fact]
    public void TryGetDataSetReturnsFalseWhenNotFound()
    {
        // Arrange
        var mockDataGateway = new Mock<IDataGateway>();
        var context = new CalculationContext(mockDataGateway.Object);

        // Act
        var found = context.TryGetDataSet<string>("Missing", out var dataset);

        // Assert
        found.ShouldBeFalse();
        dataset.ShouldBeNull();
    }

    [Fact]
    public void SetCalculationResultStoresAndRetrievesTypedResult()
    {
        // Arrange
        var mockDataGateway = new Mock<IDataGateway>();
        var context = new CalculationContext(mockDataGateway.Object);

        // Act
        context.SetCalculationResult("TotalSales", 1250.50m);

        // Assert
        context.HasCalculationResult("TotalSales").ShouldBeTrue();
        context.GetCalculationResult<decimal>("TotalSales").ShouldBe(1250.50m);
    }
}

Dependencies

• FractalDataWorks.Calculations.Abstractions: Core calculation interfaces
• FractalDataWorks.Services.Data.Abstractions: IDataGateway interface
• FractalDataWorks.Commands.Data.Abstractions: Data command interfaces
• FractalDataWorks.Results: IGenericResult<T> implementation

Relationship to Other Packages

FractalDataWorks.Calculations (this package)
├── Implements: FractalDataWorks.Calculations.Abstractions
├── Uses: FractalDataWorks.Services.Data.Abstractions
├── Integrates with: FractalDataWorks.Etl.Pipelines
└── Extended by: Custom calculation implementations

See Also

• FractalDataWorks.Calculations.Abstractions: Interfaces and contracts
• FractalDataWorks.Calculations.Aggregations: Aggregation functions TypeCollection
• FractalDataWorks.Services.Data.Abstractions: Data gateway documentation
• FractalDataWorks.Results: Result types and Railway-Oriented Programming