alt/services/worker/internal/marketdata/aggregation/monthly.go

// Package aggregation builds deterministic derived market-data bars from
// normalized worker market bars.
package aggregation

import (
	"fmt"
	"math/big"
	"sort"
	"strings"
	"time"

	"git.toki-labs.com/toki/alt/packages/domain/market"
)

const MonthlyAggregationRuleVersion = "daily_to_monthly_ohlcv_v1"

// MonthlyProvenance records the daily source range and rule used to derive one
// monthly bar. It is intentionally small enough to use as stable fixture output
// before durable aggregation metadata is introduced.
type MonthlyProvenance struct {
	InstrumentID    market.InstrumentID `json:"instrument_id"`
	SourceTimeframe market.Timeframe    `json:"source_timeframe"`
	TargetTimeframe market.Timeframe    `json:"target_timeframe"`
	TargetTimestamp time.Time           `json:"target_timestamp"`
	SourceStart     time.Time           `json:"source_start"`
	SourceEnd       time.Time           `json:"source_end"`
	SourceCount     int                 `json:"source_count"`
	RuleVersion     string              `json:"rule_version"`
}

// AggregateDailyToMonthly converts one instrument's daily bars into monthly
// OHLCV bars. Input order is irrelevant; the output is sorted by month.
func AggregateDailyToMonthly(bars []market.Bar) ([]market.Bar, []MonthlyProvenance, error) {
	if len(bars) == 0 {
		return nil, nil, nil
	}

	sorted := make([]market.Bar, len(bars))
	copy(sorted, bars)
	sort.SliceStable(sorted, func(i, j int) bool {
		return sorted[i].Timestamp.Before(sorted[j].Timestamp)
	})

	instrumentID := sorted[0].InstrumentID
	currency, err := barCurrency(sorted[0])
	if err != nil {
		return nil, nil, fmt.Errorf("monthly aggregation: bar 0: %w", err)
	}

	var out []market.Bar
	var provenance []MonthlyProvenance
	var current *monthlyBucket

	for i, bar := range sorted {
		if bar.InstrumentID != instrumentID {
			return nil, nil, fmt.Errorf("monthly aggregation: mixed instrument ids %q and %q", instrumentID, bar.InstrumentID)
		}
		if bar.Timeframe != market.TimeframeDaily {
			return nil, nil, fmt.Errorf("monthly aggregation: bar %d timeframe %q is not daily", i, bar.Timeframe)
		}
		if got, err := barCurrency(bar); err != nil {
			return nil, nil, fmt.Errorf("monthly aggregation: bar %d: %w", i, err)
		} else if got != currency {
			return nil, nil, fmt.Errorf("monthly aggregation: mixed currencies %q and %q", currency, got)
		}
		if err := validateBarDecimals(i, bar); err != nil {
			return nil, nil, err
		}

		monthStart := monthStartUTC(bar.Timestamp)
		if current == nil || !current.month.Equal(monthStart) {
			if current != nil {
				out = append(out, current.bar())
				provenance = append(provenance, current.provenance())
			}
			current = newMonthlyBucket(monthStart, bar)
			continue
		}
		current.add(bar)
	}

	if current != nil {
		out = append(out, current.bar())
		provenance = append(provenance, current.provenance())
	}
	return out, provenance, nil
}

type monthlyBucket struct {
	month       time.Time
	instrument  market.InstrumentID
	sourceStart time.Time
	sourceEnd   time.Time
	count       int

	open      market.Price
	high      market.Price
	highValue *big.Rat
	low       market.Price
	lowValue  *big.Rat
	close     market.Price
	volume    *big.Rat
}

func newMonthlyBucket(month time.Time, bar market.Bar) *monthlyBucket {
	high := mustDecimalRat(bar.High.Amount)
	low := mustDecimalRat(bar.Low.Amount)
	volume := mustDecimalRat(bar.Volume.Amount)
	return &monthlyBucket{
		month:       month,
		instrument:  bar.InstrumentID,
		sourceStart: bar.Timestamp,
		sourceEnd:   bar.Timestamp,
		count:       1,
		open:        bar.Open,
		high:        bar.High,
		highValue:   high,
		low:         bar.Low,
		lowValue:    low,
		close:       bar.Close,
		volume:      volume,
	}
}

func (b *monthlyBucket) add(bar market.Bar) {
	if high := mustDecimalRat(bar.High.Amount); high.Cmp(b.highValue) > 0 {
		b.high = bar.High
		b.highValue = high
	}
	if low := mustDecimalRat(bar.Low.Amount); low.Cmp(b.lowValue) < 0 {
		b.low = bar.Low
		b.lowValue = low
	}
	b.close = bar.Close
	b.sourceEnd = bar.Timestamp
	b.count++
	b.volume.Add(b.volume, mustDecimalRat(bar.Volume.Amount))
}

func (b *monthlyBucket) bar() market.Bar {
	return market.Bar{
		InstrumentID: b.instrument,
		Timeframe:    market.TimeframeMonthly,
		Timestamp:    b.month,
		Open:         b.open,
		High:         b.high,
		Low:          b.low,
		Close:        b.close,
		Volume:       market.Quantity{Amount: decimalValue(b.volume)},
	}
}

func (b *monthlyBucket) provenance() MonthlyProvenance {
	return MonthlyProvenance{
		InstrumentID:    b.instrument,
		SourceTimeframe: market.TimeframeDaily,
		TargetTimeframe: market.TimeframeMonthly,
		TargetTimestamp: b.month,
		SourceStart:     b.sourceStart,
		SourceEnd:       b.sourceEnd,
		SourceCount:     b.count,
		RuleVersion:     MonthlyAggregationRuleVersion,
	}
}

func monthStartUTC(ts time.Time) time.Time {
	year, month, _ := ts.Date()
	return time.Date(year, month, 1, 0, 0, 0, 0, time.UTC)
}

func barCurrency(bar market.Bar) (market.Currency, error) {
	currency := bar.Open.Currency
	if currency == "" {
		return "", fmt.Errorf("open currency is required")
	}
	for name, got := range map[string]market.Currency{
		"high":  bar.High.Currency,
		"low":   bar.Low.Currency,
		"close": bar.Close.Currency,
	} {
		if got != currency {
			return "", fmt.Errorf("%s currency %q does not match open currency %q", name, got, currency)
		}
	}
	return currency, nil
}

func validateBarDecimals(idx int, bar market.Bar) error {
	for name, value := range map[string]market.Decimal{
		"open":   bar.Open.Amount,
		"high":   bar.High.Amount,
		"low":    bar.Low.Amount,
		"close":  bar.Close.Amount,
		"volume": bar.Volume.Amount,
	} {
		if _, err := decimalRat(value); err != nil {
			return fmt.Errorf("monthly aggregation: bar %d invalid %s decimal: %w", idx, name, err)
		}
	}
	return nil
}

func mustDecimalRat(decimal market.Decimal) *big.Rat {
	rat, err := decimalRat(decimal)
	if err != nil {
		panic(err)
	}
	return rat
}

func decimalRat(decimal market.Decimal) (*big.Rat, error) {
	value := strings.TrimSpace(decimal.Value)
	if value == "" {
		return nil, fmt.Errorf("empty decimal value")
	}
	rat, ok := new(big.Rat).SetString(value)
	if !ok {
		return nil, fmt.Errorf("invalid decimal value %q", decimal.Value)
	}
	return rat, nil
}

func decimalValue(rat *big.Rat) market.Decimal {
	if rat.Sign() == 0 {
		return market.Decimal{Value: "0"}
	}
	if rat.IsInt() {
		return market.Decimal{Value: rat.Num().String()}
	}
	value := strings.TrimRight(rat.FloatString(18), "0")
	value = strings.TrimRight(value, ".")
	if value == "" || value == "-0" {
		value = "0"
	}
	return market.Decimal{Value: value}
}