|
|
|
@@ -51,13 +51,52 @@ type ReportWriter interface {
|
|
|
|
|
Write(context.Context, ReportItem) error
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
type optionals struct {
|
|
|
|
|
selector Selector
|
|
|
|
|
store Store
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func applyOptions(defaults optionals, opts []Option) optionals {
|
|
|
|
|
for _, opt := range opts {
|
|
|
|
|
opt(&defaults)
|
|
|
|
|
}
|
|
|
|
|
return defaults
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
type Option func(*optionals)
|
|
|
|
|
|
|
|
|
|
// Selector returns true if a record should be selected for processing, false otherwise.
|
|
|
|
|
type Selector func(Record) bool
|
|
|
|
|
|
|
|
|
|
func WithSelector(s Selector) Option {
|
|
|
|
|
return func(o *optionals) {
|
|
|
|
|
o.selector = s
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
type Store interface {
|
|
|
|
|
Load() (map[string]*FillerQueue, error)
|
|
|
|
|
Save(map[string]*FillerQueue) error
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func WithStore(s Store) Option {
|
|
|
|
|
return func(o *optionals) {
|
|
|
|
|
o.store = s
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// BuildReport reads records from a RecordReader and, if the record passes the Selector, it is
|
|
|
|
|
// processed into the ReportWriter.
|
|
|
|
|
func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter, sel Selector) error {
|
|
|
|
|
buys := make(map[string]*FillerQueue)
|
|
|
|
|
func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter, options ...Option) error {
|
|
|
|
|
optionals := applyOptions(optionals{
|
|
|
|
|
selector: Any(),
|
|
|
|
|
store: EphemeralStore{},
|
|
|
|
|
}, options)
|
|
|
|
|
|
|
|
|
|
buys, err := optionals.store.Load()
|
|
|
|
|
if err != nil {
|
|
|
|
|
return fmt.Errorf("loading state: %w", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
var buysCount, sellsCount int64
|
|
|
|
|
var lastTimestamp time.Time
|
|
|
|
@@ -79,6 +118,11 @@ func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter,
|
|
|
|
|
rec, err := reader.ReadRecord(ctx)
|
|
|
|
|
if err != nil {
|
|
|
|
|
if errors.Is(err, io.EOF) {
|
|
|
|
|
err = optionals.store.Save(buys)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return fmt.Errorf("saving state: %w", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
return err
|
|
|
|
@@ -98,21 +142,18 @@ func BuildReport(ctx context.Context, reader RecordReader, writer ReportWriter,
|
|
|
|
|
buys[rec.Symbol()] = buyQueue
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
err = processRecord(ctx, buyQueue, rec, sel, writer)
|
|
|
|
|
err = processRecord(ctx, buyQueue, rec, optionals.selector, writer)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return fmt.Errorf("processing record: %w", err)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// processRecord either adds buys to the queue or consumes buys from the queue when processing a
|
|
|
|
|
// sell record.
|
|
|
|
|
//
|
|
|
|
|
// NOTE: Selectors are only applied when processing sell records for performance reasons. It's much
|
|
|
|
|
// cheaper to just accumulate buys and only actually inspect any records once a sell happens. This
|
|
|
|
|
// avoids potential network requests to for every single record.
|
|
|
|
|
func processRecord(ctx context.Context, q *FillerQueue, rec Record, sel Selector, writer ReportWriter) error {
|
|
|
|
|
slog.Debug(
|
|
|
|
|
"Report: processing record",
|
|
|
|
@@ -122,6 +163,9 @@ func processRecord(ctx context.Context, q *FillerQueue, rec Record, sel Selector
|
|
|
|
|
|
|
|
|
|
switch rec.Kind() {
|
|
|
|
|
case KindBuy:
|
|
|
|
|
// Selectors are only applied when processing sell records for performance reasons. It's much
|
|
|
|
|
// cheaper to just accumulate buys and only actually inspect any records once a sell happens. This
|
|
|
|
|
// avoids potential network requests to for every single record.
|
|
|
|
|
q.Push(NewFiller(rec))
|
|
|
|
|
|
|
|
|
|
case KindSell:
|
|
|
|
@@ -142,7 +186,8 @@ func processRecord(ctx context.Context, q *FillerQueue, rec Record, sel Selector
|
|
|
|
|
return ErrInsufficientBoughtVolume
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Since we don't apply selectors while processing buys we need to apply them here.
|
|
|
|
|
// Since we don't apply selectors while processing buys we need to apply them here, befre we
|
|
|
|
|
// actually use them.
|
|
|
|
|
if !sel(buy) {
|
|
|
|
|
continue
|
|
|
|
|
}
|
|
|
|
|