Files
spark/src/DataStreamSuite/DataStreamCore/BlueData/BlueTableImmutable.cs
T
John Colvin 04200a330a feat: Add historic market data streams factory
Loads historic md from Databento files in structured paths. It is
designed to work with files downloaded via databento_client.py
2025-11-19 14:40:15 -05:00

486 lines
16 KiB
C#

#nullable enable
namespace Bts.DataStreamCore;
using BCG;
using bts.api.core;
using bts.collections;
using Spark.Core.Collections;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Linq;
// I would have named this ImmutableBlueTable, but that was already taken.
public class BlueTableImmutable : IBlueRowCollection, IImmutableRowCollection {
readonly ImmutableSortedSet<KeyValuePair<int, IBlueRow>> rows;
public BlueTableImmutable(IBlueTableSchema schema)
: this(schema, ImmutableSortedSet.Create(KeyValuePairComparer.KeyComparer<int, IBlueRow>())) { }
BlueTableImmutable(IBlueTableSchema schema, ImmutableSortedSet<KeyValuePair<int, IBlueRow>> rows) {
this.schema = schema;
this.rows = rows;
}
public int Count => rows.Count;
public bool IsSynchronized => false;
public IBlueTableSchema schema { get; }
public object SyncRoot => this;
public IBlueRow this[int id] =>
#nullable disable
rows.TryGetValue(new(id, default), out var kv)
#nullable enable
? kv.Value
: throw new KeyNotFoundException();
IBlueRow IList<IBlueRow>.this[int index] {
get => GetRow(index);
set => throw new NotSupportedException();
}
public static DataPacketBuilder CreateBuilder(IBlueTableSchema schema) =>
new(new(schema));
public static BlueTableImmutable CreateRange(IBlueTableSchema schema, IEnumerable<KeyValuePair<int, IBlueRow>> rows) =>
new(schema, ImmutableSortedSet.CreateRange(KeyValuePairComparer.KeyComparer<int, IBlueRow>(), rows));
public IBlueRow GetRow(int index) =>
rows[index].Value;
public int IndexOf(IBlueRow item) {
int id = item.Id();
if (rows.TryGetValue(new(id, item), out var rowKvp)) {
var vals = schema.columnInfo.Select(c => rowKvp.Value[c.colNo]);
var itemVals = item.schema.columnInfo.Select(c => item[c.colNo]);
if (vals.SequenceEqual(itemVals)) {
return rows.IndexOf(rowKvp);
}
}
return -1;
}
public bool containsID(int id) =>
#nullable disable
rows.Contains(new(id, default));
#nullable enable
public void CopyTo(Array array, int index) {
foreach (var kv in rows) {
array.SetValue(kv.Value, index++);
}
}
public IList<object?> GetColumn(int index) =>
rows.WrapSelector(kv => kv.Value[index], null);
public IEnumerator<IBlueRow> GetEnumerator() {
foreach (var kv in rows) {
yield return kv.Value;
}
}
public Builder ToBuilder() =>
new(this);
public DataPacketBuilder ToDataPacketBuilder() =>
new(this);
public class Builder : IEnumerable<IBlueRow> {
readonly IBlueTableSchema schema;
readonly ImmutableSortedSetDictionary<int, IBlueRow>.Builder rowMap;
public Builder(BlueTableImmutable source) {
schema = source.schema;
rowMap = ImmutableSortedSetDictionary.CreateBuilder(source.rows.ToBuilder());
}
public void AddRange(IEnumerable<KeyValuePair<int, IBlueRow>> rows) =>
rowMap.AddRange(rows);
public void AddOrUpdate(IBlueRow row) =>
AddOrUpdate(row.Id(), row);
public void AddOrUpdate(int id, IBlueRow row) =>
rowMap[id] = row;
public bool Remove(int id) =>
rowMap.Remove(id);
public BlueTableImmutable ToImmutable() =>
new(schema, rowMap.ToImmutable().Source);
public IEnumerator<IBlueRow> GetEnumerator() =>
rowMap.Values.GetEnumerator();
IEnumerator IEnumerable.GetEnumerator() =>
GetEnumerator();
}
public class DataPacketBuilder {
readonly IBlueTableSchema schema;
readonly ImmutableSortedSet<KeyValuePair<int, IBlueRow>>.Builder table;
readonly ImmutableSortedSet<KeyValuePair<int, IBlueRow>>.Builder added =
ImmutableSortedSet.CreateBuilder(KeyValuePairComparer.KeyComparer<int, IBlueRow>());
readonly ImmutableSortedSet<KeyValuePair<int, IBlueRow>>.Builder removed =
ImmutableSortedSet.CreateBuilder(KeyValuePairComparer.KeyComparer<int, IBlueRow>());
public DataPacketBuilder(BlueTableImmutable table) {
schema = table.schema;
this.table = table.rows.ToBuilder();
}
public void AddOrUpdate(IBlueRow row, IEqualityComparer<IBlueRow>? comparer = null) {
var id = row.Id();
#nullable disable
if (table.TryGetValue(new(id, default), out var orig)) {
#nullable enable
if (comparer?.Equals(row, orig.Value) ?? false) {
// No op: This row was already in the table (according to 'comparer').
// Remove it from 'added' and 'removed'.
added.Remove(new(id, row));
removed.Remove(new(id, orig.Value));
return;
}
// Change op: A row with this id was already in the table,
// but its values have changed (according to 'comparer').
removed.Remove(new(id, orig.Value));
removed.Add(new(id, orig.Value));
}
added.Remove(new(id, row));
added.Add(new(id, row));
}
public void Remove(int id) {
#nullable disable
added.Remove(new(id, default));
if (table.TryGetValue(new(id, default), out var orig)) {
#nullable enable
removed.Remove(new(id, orig.Value));
removed.Add(new(id, orig.Value));
}
}
public void Clear() {
added.Clear();
removed.Clear();
removed.AddRange(table);
}
public void Reset() {
added.Clear();
removed.Clear();
}
public DataPacket ToDataPacket(bool reset = true) =>
ToDataPacket(out _, reset);
public DataPacket ToDataPacket(out BlueTableImmutable newRows, bool reset = true) {
table.RemoveRange(removed);
table.AddRange(added);
newRows = new BlueTableImmutable(schema, table.ToImmutable());
var addedRows = new BlueTableImmutable(schema, added.ToImmutable());
var removedRows = new BlueTableImmutable(schema, removed.ToImmutable());
if (reset) {
added.Clear();
removed.Clear();
}
return new(newRows, addedRows, removedRows);
}
public BlueTableImmutable ToImmutable() =>
new(schema, table.ToImmutable());
}
}
// public class BlueTableImmutable<T> : IBlueRowCollection<T> where T : IBlueRow {
// readonly ImmutableDictionary<int, T> rows;
// public BlueTableImmutable(IBlueTableSchema schema)
// : this(schema, ImmutableDictionary<int, T>.Empty) { }
// public BlueTableImmutable(IBlueTableSchema schema, ImmutableDictionary<int, T> rows) {
// this.schema = schema;
// this.rows = rows;
// }
// public int Count => rows.Count;
// public bool IsSynchronized => false;
// public IBlueTableSchema schema { get; }
// public object SyncRoot => this;
// public T this[int id] => rows[id];
// public bool containsID(int id) =>
// rows.ContainsKey(id);
// public void CopyTo(Array array, int index) {
// foreach (var r in rows.Values) {
// array.SetValue(r, index++);
// }
// }
// public IEnumerator<T> GetEnumerator() =>
// rows.Values.GetEnumerator();
// IEnumerator IEnumerable.GetEnumerator() =>
// GetEnumerator();
// public DataPacketBuilder ToDataPacketBuilder() =>
// new(this);
// public class DataPacketBuilder {
// readonly IBlueTableSchema schema;
// readonly ImmutableDictionary<int, T>.Builder table;
// readonly ImmutableDictionary<int, T>.Builder added =
// ImmutableDictionary<int, T>.Empty.ToBuilder();
// readonly ImmutableDictionary<int, T>.Builder removed =
// ImmutableDictionary<int, T>.Empty.ToBuilder();
// public DataPacketBuilder(BlueTableImmutable<T> table) {
// schema = table.schema;
// this.table = table.rows.ToBuilder();
// }
// public void AddOrUpdate(T row, IEqualityComparer<IBlueRow>? comparer = null) {
// var id = row.Id();
// if (table.TryGetValue(id, out var orig)) {
// if (comparer?.Equals(row, orig) ?? false) {
// // No op: This row was already in the table (according to 'comparer').
// // Remove it from 'added' and 'removed'.
// added.Remove(id);
// removed.Remove(id);
// return;
// }
// // Change op: A row with this id was already in the table,
// // but its values have changed (according to 'comparer').
// removed[id] = orig;
// }
// added[id] = row;
// }
// public void Remove(int id) {
// added.Remove(id);
// if (table.TryGetValue(id, out var orig)) {
// removed[id] = orig;
// }
// }
// public DataPacket ToDataPacket(bool reset = true) {
// table.RemoveRange(removed.Keys);
// table.AddRange(added);
// var newRows = new BlueTableImmutable<T>(schema, table.ToImmutable());
// var addedRows = new BlueTableImmutable<T>(schema, added.ToImmutable());
// var removedRows = new BlueTableImmutable<T>(schema, removed.ToImmutable());
// if (reset) {
// added.Clear();
// removed.Clear();
// }
// return new(newRows, addedRows, removedRows);
// }
// public IDataPacket<T> ToDataPacketT(bool reset = true) {
// table.RemoveRange(removed.Keys);
// table.AddRange(added);
// var newRows = new BlueTableImmutable<T>(schema, table.ToImmutable());
// var addedRows = new BlueTableImmutable<T>(schema, added.ToImmutable());
// var removedRows = new BlueTableImmutable<T>(schema, removed.ToImmutable());
// if (reset) {
// added.Clear();
// removed.Clear();
// }
// return new DataPacket<T>(newRows, addedRows, removedRows);
// }
// public IDataPacket<T> ToDataPacketT(
// out BlueTableImmutable<T> newRows,
// out BlueTableImmutable<T> addedRows,
// out BlueTableImmutable<T> removedRows,
// bool reset = true
// ) {
// table.RemoveRange(removed.Keys);
// table.AddRange(added);
// newRows = new BlueTableImmutable<T>(schema, table.ToImmutable());
// addedRows = new BlueTableImmutable<T>(schema, added.ToImmutable());
// removedRows = new BlueTableImmutable<T>(schema, removed.ToImmutable());
// if (reset) {
// added.Clear();
// removed.Clear();
// }
// return new DataPacket<T>(newRows, addedRows, removedRows);
// }
// public BlueTableImmutable<T> ToImmutable() =>
// new(schema, table.ToImmutable());
// }
// }
public class BlueTableImmutableAdapter<T> : IBlueRowCollection {
readonly BlueTableSchemaAdapter<T> schemaAdapter;
readonly ImmutableSortedSet<KeyValuePair<int, T>> rows;
public BlueTableImmutableAdapter(BlueTableSchemaAdapter<T> schemaAdapter)
: this(schemaAdapter, ImmutableSortedSet.Create(KeyValuePairComparer.KeyComparer<int, T>())) { }
BlueTableImmutableAdapter(BlueTableSchemaAdapter<T> schemaAdapter, ImmutableSortedSet<KeyValuePair<int, T>> rows) {
this.schemaAdapter = schemaAdapter;
this.rows = rows;
}
public int Count => rows.Count;
public bool IsSynchronized => false;
public IBlueTableSchema schema => schemaAdapter.Schema;
public object SyncRoot => this;
public IBlueRow this[int id] =>
new BlueRowAdapter<T>(schemaAdapter, rows[id].Value);
public bool containsID(int id) =>
#nullable disable
rows.Contains(new(id, default));
#nullable enable
public IList<object?> GetColumn(int index) =>
rows.WrapSelector(kv => schemaAdapter.GetBlueRow(kv.Value)[index], null);
public IBlueRow GetRow(int index) =>
new BlueRowAdapter<T>(schemaAdapter, rows[index].Value);
public void CopyTo(Array array, int index) {
foreach (var kv in rows) {
array.SetValue(new BlueRowAdapter<T>(schemaAdapter, kv.Value), index++);
}
}
public int IndexOf(IBlueRow item) {
int id = item.Id();
#nullable disable
int index = rows.IndexOf(new(id, default));
if (index != -1) {
var r = this[index];
if (BlueRowEqualityComparer.Equals(r, item)) {
return index;
}
}
#nullable enable
return -1;
}
public IEnumerator<IBlueRow> GetEnumerator() {
foreach (var kv in rows) {
yield return new BlueRowAdapter<T>(schemaAdapter, kv.Value);
}
}
public Builder ToBuilder() =>
new(this);
public DataPacketBuilder ToDataPacketBuilder() =>
new(this);
public class Builder : IEnumerable<IBlueRow> {
readonly BlueTableSchemaAdapter<T> schemaAdapter;
readonly ImmutableSortedSetDictionary<int, T>.Builder rowMap;
public Builder(BlueTableImmutableAdapter<T> source) {
schemaAdapter = source.schemaAdapter;
rowMap = ImmutableSortedSetDictionary.CreateBuilder(source.rows.ToBuilder());
}
public void AddRange(IEnumerable<KeyValuePair<int, T>> rows) =>
rowMap.AddRange(rows);
public void AddOrUpdate(T row) =>
AddOrUpdate((int)schemaAdapter["id"][row]!, row);
public void AddOrUpdate(int id, T row) =>
rowMap[id] = row;
public bool Remove(int id) =>
rowMap.Remove(id);
public BlueTableImmutableAdapter<T> ToImmutable() =>
new(schemaAdapter, rowMap.ToImmutable().Source);
public IEnumerator<IBlueRow> GetEnumerator() {
foreach (var r in rowMap.Values) {
yield return schemaAdapter.GetBlueRow(r);
}
}
IEnumerator IEnumerable.GetEnumerator() =>
GetEnumerator();
}
public class DataPacketBuilder {
readonly BlueTableSchemaAdapter<T> schemaAdapter;
readonly ImmutableSortedSet<KeyValuePair<int, T>>.Builder table =
ImmutableSortedSet.CreateBuilder(KeyValuePairComparer.KeyComparer<int, T>());
readonly ImmutableSortedSet<KeyValuePair<int, T>>.Builder added =
ImmutableSortedSet.CreateBuilder(KeyValuePairComparer.KeyComparer<int, T>());
readonly ImmutableSortedSet<KeyValuePair<int, T>>.Builder removed =
ImmutableSortedSet.CreateBuilder(KeyValuePairComparer.KeyComparer<int, T>());
public DataPacketBuilder(BlueTableImmutableAdapter<T> table) {
schemaAdapter = table.schemaAdapter;
this.table = table.rows.ToBuilder();
}
public void AddOrUpdate(T rowData, IEqualityComparer<IBlueRow>? comparer = null) {
var row = new BlueRowAdapter<T>(schemaAdapter, rowData);
var id = row.Id();
#nullable disable
if (table.TryGetValue(new(id, default), out var origData)) {
#nullable enable
var orig = new BlueRowAdapter<T>(schemaAdapter, origData.Value);
if (comparer?.Equals(row, orig) ?? false) {
// No op: This row was already in the table (according to 'comparer').
// Remove it from 'added' and 'removed'.
added.Remove(origData);
removed.Remove(origData);
return;
}
removed.Add(origData);
}
added.Remove(new(id, default!));
added.Add(new(id, rowData));
}
public void Remove(int id) {
#nullable disable
added.Remove(new(id, default));
if (table.TryGetValue(new(id, default), out var orig)) {
#nullable enable
removed.Remove(orig);
removed.Add(orig);
}
}
public void Update<TKey>(TableUpdate<T, TKey> tableUpdate) where TKey : IEquatable<TKey> {
foreach (var r in tableUpdate.removed.Concat(tableUpdate.changes.Old())) {
Remove(new BlueRowAdapter<T>(schemaAdapter, r).Id());
}
foreach (var r in tableUpdate.changed.Concat(tableUpdate.added)) {
AddOrUpdate(r);
}
}
public BlueTableImmutableAdapter<T> ToImmutable(out IDataPacket dp, bool reset = true) {
table.RemoveRange(removed);
table.AddRange(added);
var newRows = new BlueTableImmutableAdapter<T>(schemaAdapter, table.ToImmutable());
var addedRows = new BlueTableImmutableAdapter<T>(schemaAdapter, added.ToImmutable());
var removedRows = new BlueTableImmutableAdapter<T>(schemaAdapter, removed.ToImmutable());
if (reset) {
added.Clear();
removed.Clear();
}
dp = new DataPacket(newRows, addedRows, removedRows);
return newRows;
}
}
}