Compare commits

...
Author SHA1 Message Date
John Colvin 2e78af1756 fix build: Upgrade IronPython to 3.4.2
All csprojs need to be on same version (Spark.csproj was on 3.4.1, while
DataStream.csproj was on 3.4.2)
2025-06-16 11:08:50 -04:00
Taha 70fae06a30 BlueOrb streams and MLP 2025-06-08 21:52:38 -04:00
John Colvin 07b59559a2 dev qol: Reduce md log level in config profile 2025-06-03 02:23:59 +00:00
John Colvin cd1709563b fix: Generate reject exe reports from ack at binance
This only applies to "new" order acks. In all other cases, exe reports
are generated from the order update stream. New order rejects are the
exception because they're not visible in the stream. Presumably, they're
never actually entered into binance's order system (they're "rejected"
before that happens).
2025-06-03 02:21:18 +00:00
arham 94f22ddf40 Update: changed url for okx ws exe in config 2025-05-30 12:59:26 +00:00
arham 7f6c08f956 Fix: positions now go through dispatch incoming 2025-05-29 16:26:05 +00:00
arham 1d737cdf82 Fix: change time to 5s for pos update okx 2025-05-23 16:12:12 +00:00
arham b645192355 Add: OKX position turned on 2025-05-23 15:33:11 +00:00
45 changed files with 2756 additions and 609 deletions

No files matched your search

@@ -0,0 +1,49 @@
using System;
namespace BCG {
[DataStreamUserRespresentation("count")]
public class CountStream : ViewStream {
private DataPacket myLastFiredDataPacket;
private IBlueTableSchema mySchema = new BlueTableSchema("count",
new ColumnInfo("id", 0, typeof(int)),
new ColumnInfo("count", 1, typeof(int)));
private int myId = 0;
private ViewStream myVS;
public CountStream(ViewStream vs) {
myVS = vs;
}
public CountStream(string tableName)
: this(new TableStream(tableName)) { }
protected override void myOnFirstSubscriber() {
myVS.onChange += new ChangeDelegate(myHandleChanges);
}
protected override void myOnLastSubscriber() {
myVS.onChange -= new ChangeDelegate(myHandleChanges);
}
private void myHandleChanges(object newValue) {
myFireEvent(
myLastFiredDataPacket =
myLastFiredDataPacket == null ?
new DataPacket(myGetRows(newValue, ++myId), myGetRows(newValue, myId),
new BlueTable(mySchema))
:
new DataPacket(myGetRows(newValue, ++myId), myGetRows(newValue, myId),
myLastFiredDataPacket.newRows)
);
}
private IBlueRowCollection myGetRows(object dp, int id) {
BlueTable retVal = new BlueTable(mySchema);
retVal.addRow
(new BlueRow(mySchema, id, (dp as DataPacket).newRows.Count));
return retVal;
}
}
}
@@ -0,0 +1,38 @@
namespace bts.BlueOrb {
using bts;
using BCG;
using System;
[DataStreamUserRespresentation("DateTimeExpanderStream")]
public class DateTimeExpanstionStream : ViewStream {
public DateTimeExpanstionStream(ViewStream parent_stream, string inp_col) {
Func<IBlueTableSchema, IBlueTableSchema> get_schema = (s) => {
IBlueTableSchema rv = s.add_column(System.TypeCode.DateTime, $"{inp_col}_date");
rv = rv.add_column(System.TypeCode.String, $"{inp_col}_day_of_week");
rv = rv.add_column(System.TypeCode.Int32, $"{inp_col}_day_of_month");
rv = rv.add_column(System.TypeCode.Int32, $"{inp_col}_month");
rv = rv.add_column(System.TypeCode.Int32, $"{inp_col}_hour");
rv = rv.add_column(System.TypeCode.Int32, $"{inp_col}_minute");
rv = rv.add_column(System.TypeCode.Int32, $"{inp_col}_seconds_of_day");
return rv;
};
ViewStream rs = new MapStream(parent_stream,
s => get_schema(s),
r => {
DateTime dt = r.get_datetime(inp_col);
IBlueTableSchema s = get_schema(r.schema);
MutableBlueRow rv = new MutableBlueRow(s, r);
rv.setValue($"{inp_col}_date", dt.Date);
rv.setValue($"{inp_col}_day_of_week", dt.DayOfWeek.ToString());
rv.setValue($"{inp_col}_day_of_month", dt.Day);
rv.setValue($"{inp_col}_month", dt.Date.Month);
rv.setValue($"{inp_col}_hour", dt.Hour);
rv.setValue($"{inp_col}_minute", dt.Minute);
rv.setValue($"{inp_col}_seconds_of_day", (dt - dt.Date).TotalSeconds);
return rv;
});
mySetRealSource(rs);
}
}
}
@@ -0,0 +1,158 @@
namespace bts.BlueOrb {
using BCG;
using System;
using System.Text.RegularExpressions;
public class RegExpFilter : IStreamFilter {
private Regex[] myRegExps;
private string[] myCols;
public RegExpFilter(string col, string regExp)
: this(new string[] { col }, new string[] { regExp }) { }
public RegExpFilter(string[] cols, string[] regExps) {
myCols = cols;
myRegExps = new Regex[regExps.Length];
for (int i = 0; i < regExps.Length; i++)
myRegExps[i] = new Regex(regExps[i]);
}
#region IStreamFilter Members
public bool filter(IBlueRow dr) {
bool rv = true;
for (int i = 0; i < myCols.Length; i++) {
rv &= myRegExps[i].IsMatch(dr[myCols[i]] is DBNull ? "" : dr[myCols[i]].ToString());
if (!rv)
return rv;
}
return rv;
}
#endregion
}
//[DataStreamUserReprsentation("FilterStream")]
public class FilterStream : ViewStream {
private BlueTable myLastFiredRows;
private IBlueRowCollection myLastReceivedRows;
private IStreamFilter myFilter;
private ViewStream myParentStream;
private string myFilterString;
private bool myUseNewFilter = false;// Configuration.get<DataStreamConfig>().UseNewFilter;
#region constructors
public delegate bool OnFilter(IBlueRow r);
private class OnFilterWrapper : IStreamFilter {
public OnFilterWrapper(OnFilter f) { this.f = f; }
public bool filter(IBlueRow dr) { return f(dr); }
private OnFilter f;
}
public FilterStream(ViewStream vs, OnFilter f)
: this(vs, new OnFilterWrapper(f)) { }
public FilterStream(ViewStream vs, IStreamFilter filter) {
{
myParentStream = vs;
myFilter = filter;
myFilterString = filter.ToString();
if (filter is IDynamicStreamFilter)
(filter as IDynamicStreamFilter).onFilterChange += new FilterChangeDelegate(myHandleFilterChange);
}
}
public FilterStream(ViewStream vs, string filterExpression)
: this(vs, new BooleanExpressionFilter(filterExpression)) {
myFilterString = filterExpression;
}
public FilterStream(ViewStream vs, string col, string regExp)
: this(vs, new RegExpFilter(col, regExp)) {
myFilterString = col + " , " + regExp;
}
public FilterStream(ViewStream vs)
: this(vs, new BinaryFilter(true)) {
myFilterString = true.ToString();
}
#endregion
#region overridden methods
protected override void myOnFirstSubscriber() { myParentStream.onChange += new ChangeDelegate(myHandleCallbacks); }
protected override void myOnLastSubscriber() {
myLastFiredRows = null;
myParentStream.onChange -= new ChangeDelegate(myHandleCallbacks);
}
protected override void myFireEvent(object o) {
DataPacket dp = o as DataPacket;
if (myLastValue == null || dp.addedRows.Count + dp.deletedRows.Count > 0)
base.myFireEvent(o);
}
#endregion
private void myHandleCallbacks(object newValue) {
DataPacket retVal;
DataPacket arrivedPacket = newValue as DataPacket;
myLastReceivedRows = arrivedPacket.newRows;
if (myLastFiredRows != null) {
retVal = new DataPacket(myLastFiredRows, new BlueTable(myLastFiredRows.schema),
new BlueTable(myLastFiredRows.schema));
foreach (IBlueRow r in arrivedPacket.deletedRows)
if (myLastFiredRows.containsID((int)r["id"])) {
myLastFiredRows.deleteRow(r);
(retVal.deletedRows as BlueTable).addRow(r);
}
foreach (IBlueRow r in arrivedPacket.addedRows)
if (myFilter.filter(r)) {
myLastFiredRows.addRow(r);
(retVal.addedRows as BlueTable).addRow(r);
}
} else {
myLastFiredRows = new BlueTable(arrivedPacket.schema);
retVal = new DataPacket(myLastFiredRows, myLastFiredRows, new BlueTable(myLastFiredRows.schema));
foreach (IBlueRow r in arrivedPacket.newRows)
if (myFilter.filter(r))
myLastFiredRows.addRow(r);
}
myFireEvent(retVal);
}
public IStreamFilter filter {
set {
{
myFilter = value;
if (numberOfSubscribers > 0) {
if (myLastReceivedRows != null) {
BlueTable nRows = new BlueTable(myLastReceivedRows.schema);
foreach (IBlueRow r in myLastReceivedRows)
if (myFilter.filter(r))
nRows.addRow(r);
IBlueRowCollection dRows = new BlueTable(nRows.schema),
aRows = new BlueTable(nRows.schema);
BlueDataUtility.computeDifferences(nRows, myLastFiredRows, out dRows, out aRows);
myLastFiredRows = nRows;
myFireEvent(new DataPacket(nRows, aRows, dRows));
}
}
}
}
get {
return myFilter;
}
}
//public string filterString { set { filter = new BooleanExpressionFilter(value); } }
public override string ToString() {
return GetType().ToString() + " " + myFilterString;
}
private void myHandleFilterChange() {
filter = myFilter;
}
}
}
@@ -0,0 +1,121 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text.Json;
using bts;
//using bts.SilverPine;
using BCG;
namespace bts.BlueOrb {
[DataStreamUserRespresentation("GroupByStream")]
public class GroupByStream : ViewStream {
public interface INonIncGroupByEvaluator {
// grouped_rows
IBlueRowCollection evaluate(IDictionary<int, List<IBlueRow>> grouped_rows);
string[] get_group_by_cols();
}
private class DelNonIncGroupByEvaluator : INonIncGroupByEvaluator {
Func<string[]> _groupby_cols_func;
Func<List<IBlueRow>, List<IBlueRow>> _evaluate_func;
public DelNonIncGroupByEvaluator(Func<string[]> groupby_cols_func,
Func<List<IBlueRow>, List<IBlueRow>> evaluate_func) {
_groupby_cols_func = groupby_cols_func;
_evaluate_func = evaluate_func;
}
public IBlueRowCollection evaluate(IDictionary<int, List<IBlueRow>> grouped_rows) {
BlueTable rv = null;
foreach (var kv in grouped_rows) {
List<IBlueRow> g_rows = kv.Value;
List<IBlueRow> out_rows = _evaluate_func.Invoke(g_rows);
if (out_rows != null && out_rows.Count > 0) {
rv = rv ?? new BlueTable(out_rows.ElementAt(0).schema);
}
rv.addRows(out_rows);
}
return rv;
}
public string[] get_group_by_cols() {
return _groupby_cols_func();
}
}
private ViewStream _main_stream;
private INonIncGroupByEvaluator _groupby_func;
private BlueRowHasher _row_hasher;
private string[] _groupby_cols;
private Stopwatch _timer = new();
private int _n_total_fires = 0, _n_expensive_fires = 0;
public GroupByStream(ViewStream main_stream, INonIncGroupByEvaluator groupby_func) {
_main_stream = main_stream;
_groupby_func = groupby_func;
_groupby_cols = _groupby_func.get_group_by_cols();
_row_hasher = new BlueRowHasher(_groupby_cols == null ? 0 : _groupby_cols.Length);
}
public GroupByStream(ViewStream main_stream, Func<string[]> groupby_cols_func,
Func<List<IBlueRow>, List<IBlueRow>> evaluate_func)
: this(main_stream, new DelNonIncGroupByEvaluator(groupby_cols_func, evaluate_func)) { }
public GroupByStream(ViewStream parent_str, string groupby_col, string standard_group_by_text)
: this(parent_str, groupby_col, standard_group_by_text, String.Empty) { }
// this is the main constructor that does everything
public GroupByStream(ViewStream parent_str, string groupby_col, string standard_group_by_ops, string optional_sort_col) {
MultiRowOpHelper std_group_by = MultiRowOpHelper.parse_ops(standard_group_by_ops);
Func<string[]> groupby_cols_func = () => {
return String.IsNullOrEmpty(groupby_col) ? null : new string[] { groupby_col };
};
Func<List<IBlueRow>, List<IBlueRow>> evaluate_func = (group_rows) => {
if (!String.IsNullOrEmpty(optional_sort_col))
group_rows.Sort((r1, r2) => BlueDataUtility.compare_rows(r2, r1, optional_sort_col));
// this si where the group by aggregation happens
IBlueRow rv = MultiRowOpHelper.group_by_compute(group_rows, std_group_by);
return new IBlueRow[] { rv }.ToList();
};
ViewStream rs = new GroupByStream(parent_str, groupby_cols_func, evaluate_func);
mySetRealSource(rs);
}
public GroupByStream(ViewStream parent_str, string standard_group_by_text) :
this(parent_str, String.Empty, standard_group_by_text) {
}
protected override void myOnFirstSubscriber() { _main_stream.onChange += _main_stream_onchange; }
protected override void myOnLastSubscriber() { _main_stream.onChange -= _main_stream_onchange; }
private void _main_stream_onchange(object nv) {
DataPacket dp = nv as DataPacket;
///grouping together
_timer.Restart();
IDictionary<int, List<IBlueRow>> grouped_rows = (_groupby_cols == null || _groupby_cols.Length == 0) ?
new Dictionary<int, List<IBlueRow>>() { { 0, dp.newRows.to_list() } } :
BlueDataUtility.group_together(dp.newRows, _groupby_cols, ref _row_hasher);
_timer.Stop();
var groupElapsedMs = _timer.ElapsedMilliseconds;
if (groupElapsedMs >= 50) {
Log.DataFlow.info($"Time to group {dp.newRows.Count} rows into {grouped_rows.Count} groups: {_timer.ElapsedMilliseconds}ms");
}
// Log.log(bts.Log.Level.Info, bts.log.Subsystem.gen, String.Format("Time to group {0} rows into {1} groups = {2}", dp.newRows.Count, grouped_rows.Count, _timer.lastIntervalTime));
/// reducing
_timer.Restart();
IBlueRowCollection total_state = _groupby_func.evaluate(grouped_rows);
_timer.Stop();
var applyElapsedMs = _timer.ElapsedMilliseconds;
_n_total_fires++;
if (applyElapsedMs >= 50) {
_n_expensive_fires++;
Log.DataFlow.info(
$"Time to evaluate {grouped_rows.Count} groups {_timer.ElapsedMilliseconds}ms, ratio {_n_expensive_fires}/{_n_total_fires}");
}
// TODO,
if (total_state != null)
myFireEvent(new DataPacket(total_state, total_state, myLastValue == null ? new BlueTable(total_state.schema) : (myLastValue as DataPacket).newRows));
}
}
}
@@ -0,0 +1,17 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using BCG;
namespace bts.BlueOrb {
public interface INonIncJoinInlineMapper
{
IBlueTableSchema get_output_schema(IBlueTableSchema schema);
IBlueRowCollection map_rows(IBlueRowCollection rows);
event MapperChangeDelegate onMapperChange;
}
}
@@ -0,0 +1,152 @@
using bts;
using bts.collections;
using mbts;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Security.Cryptography;
using BCG;
namespace bts.BlueOrb {
internal class NonIncJoinBookkeeper : IDisposable {
public Action on_new_data;
private readonly NonIncJoinSpec _join_spec;
private readonly int _join_position;
private readonly int _start_index_in_output;
private readonly BlueRowHasher _row_hasher;
private readonly bool _is_left_most;
private readonly bool _is_tag_along;
private int _n_total_fires = 0, _n_expensive_fires = 0;
private IBlueTableSchema _schema = null;
private IDictionary<int, IBlueRow> _row_map = new Dictionary<int, IBlueRow>();
private List<Pair<int, IBlueRow>> _row_list = new List<Pair<int, IBlueRow>>();
private IBlueRow _solitary_row = null;
private int[] _join_col_indices = null,
_data_col_indices = null;
public NonIncJoinBookkeeper(NonIncJoinSpec the_join_spec, int join_position, int start_index_in_output, BlueRowHasher row_hasher) {
_join_spec = the_join_spec;
_join_position = join_position;
_start_index_in_output = start_index_in_output;
_row_hasher = row_hasher;
_is_left_most = _join_position == 0;
_is_tag_along = _join_spec.join_col_names.Length == 0;
}
private bool _filled_missing_data = false;
private void _fill_missing_data(DataPacket dp) {
if (_filled_missing_data)
return;
Func<string[], bool> null_or_empty = (str_arr) => {
if (str_arr == null || str_arr.Length == 0)
return true;
foreach (string s in str_arr) {
if (!String.IsNullOrEmpty(s)) {
return false;
}
}
return true;
};
Func<string[], string[]> remove_bad_cols = (col_names) => {
HashSet<string> schema_col_map = new HashSet<string>(dp.schema.columnNames);
List<string> cols_list = new List<string>(col_names);
cols_list.RemoveWhere(c => !schema_col_map.Contains(c));
return cols_list.ToArray();
};
if (null_or_empty(_join_spec.data_col_names)) {
_join_spec.data_col_names = dp.schema.columnNames;
}
_join_spec.data_col_names = remove_bad_cols(_join_spec.data_col_names);
if (null_or_empty(_join_spec.data_col_names)) {
_join_spec.data_col_names = dp.schema.columnNames;
}
if (null_or_empty(_join_spec.projected_col_names)) {
_join_spec.projected_col_names = _join_spec.data_col_names;
}
}
public void init() {
_join_spec.stream.onChange += _on_stream_cb;
}
public void _on_stream_cb(object dp_obj) {
//Log.Gen.info($"NonIncJoinBookkeeper._on_stream_cb dp_obj: {dp_obj}");
var timer = Stopwatch.StartNew();
DataPacket dp = (DataPacket)dp_obj;
_fill_missing_data(dp);
if (_schema == null) {
_schema = dp.schema;
_join_col_indices = BlueDataUtility.get_column_indices(_schema, _join_spec.join_col_names);
_data_col_indices = BlueDataUtility.get_column_indices(_schema, _join_spec.data_col_names);
}
if (_is_left_most) {
_row_list.Clear();
foreach (IBlueRow r in dp.newRows)
_row_list.Add(new Pair<int, IBlueRow>(_row_hasher.get_hash(r, _join_col_indices), r));
} else {
if (_is_tag_along)
_solitary_row = BlueDataUtility.extractRow(dp.newRows, 0);
else {
_row_map.Clear();
foreach (IBlueRow r in dp.newRows)
_row_map[_row_hasher.get_hash(r, _join_col_indices)] = r;
}
}
timer.Stop();
_n_total_fires++;
if (timer.ElapsedMilliseconds >= 50) {
_n_expensive_fires++;
Log.Gen.info($"NonIncJoinBookkeepingData._on_stream_cb {dp.newRows.Count} rows {_join_col_indices.Length} "
+ $"key cols {_join_position} join pos time {timer.Elapsed} ratio {_n_expensive_fires}/{_n_total_fires}");
}
if (on_new_data != null)
on_new_data();
}
public bool is_ready { get { return _schema != null; } }
public void constribute_cols_to_schema(List<ColumnInfo> cis) {
for (int i = 0; i < _join_spec.data_col_names.Length; i++)
cis.Add(new ColumnInfo(_join_spec.projected_col_names[i], cis.Count, _schema[_join_spec.data_col_names[i]].colType));
}
public List<Pair<int, IBlueRow>> get_list() { return _row_list; }
public void copy_master_row_to_output_row(int row_index, object[] output_row) {
IBlueRow this_row = _row_list[row_index].second;
BlueDataUtility.copy_to_row(this_row, _data_col_indices, output_row, _start_index_in_output);
}
public void copy_matching_row_to_output_row(int row_hash, object[] output_row) {
if (_is_tag_along) {
if (_solitary_row != null)
BlueDataUtility.copy_to_row(_solitary_row, _data_col_indices, output_row, _start_index_in_output);
else
copy_dbnull_to_output_row(output_row);
} else {
IBlueRow this_row = null;
if (_row_map.TryGetValue(row_hash, out this_row))
BlueDataUtility.copy_to_row(this_row, _data_col_indices, output_row, _start_index_in_output);
else
copy_dbnull_to_output_row(output_row);
}
}
private void copy_dbnull_to_output_row(object[] output_row) {
for (int i = 0; i < _data_col_indices.Length; i++)
output_row[i + _start_index_in_output] = DBNull.Value;
}
#region IDisposable
public void Dispose() { _join_spec.stream.onChange -= _on_stream_cb; }
#endregion
}
}
@@ -0,0 +1,67 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using BCG;
using mbts;
namespace bts.BlueOrb {
// this a collection of spec, theres no joining logic in here
public class NonIncJoinSpec {
public ViewStream stream;
// join col names are the names on which the streams are joined (for equality),
// data col names are the names in the original stream
// projected col names are their new names, ie if its x:l_x, then x is in the
// data_col_names and l_x in projected cols
public string[] join_col_names, data_col_names, projected_col_names;
public NonIncJoinSpec() { }
public NonIncJoinSpec(ViewStream the_stream, string[] the_join_col_names, string[] the_data_col_names)
: this(the_stream, the_join_col_names, the_data_col_names, the_data_col_names) { }
public NonIncJoinSpec(ViewStream the_stream, string[] the_join_col_names, string[] the_data_col_names, string[] the_projected_col_names) {
stream = the_stream;
join_col_names = the_join_col_names;
data_col_names = the_data_col_names;
projected_col_names = the_projected_col_names;
}
public static NonIncJoinSpec[] from_triplet(params Triplet<ViewStream, string, string>[] join_specs) {
List<NonIncJoinSpec> rv = new List<NonIncJoinSpec>();
foreach (Triplet<ViewStream, string, string> r in join_specs) {
string[] keys = String.IsNullOrEmpty(r.second) ? new string[] { } : r.second.Split(' ');
string[] visible_cols = r.third.Split(' ');
List<string> data_col_names = new List<string>(), mapped_col_names = new List<string>();
foreach (string s in visible_cols) {
string[] toks = s.Split(':');
if (toks.Length == 1) {
data_col_names.Add(toks[0]);
mapped_col_names.Add(toks[0]);
} else if (toks.Length == 2) {
data_col_names.Add(toks[0]);
mapped_col_names.Add(toks[1]);
}
}
rv.Add(new NonIncJoinSpec(r.first, keys, data_col_names.ToArray(), mapped_col_names.ToArray()));
}
return rv.ToArray();
}
public override string ToString() {
Func<string[], string> arr_txt = (a) => {
string rv = "";
foreach (string s in a) {
rv += "{s} ";
}
rv = rv.Trim();
return rv;
};
string rv = $"stream: {stream} join_col_names: {arr_txt(join_col_names)} data_col_names: {arr_txt(data_col_names)} projected_col_names: {arr_txt(projected_col_names)}";
return rv;
}
}
}
@@ -0,0 +1,165 @@
using bts;
using mbts;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using BCG;
namespace bts.BlueOrb {
[DataStreamUserRespresentation("JoinStream")]
public class NonIncJoinStream : ViewStream, BCG.Utilities.General.IWorker {
private readonly NonIncJoinSpec[] _join_specs;
private readonly INonIncJoinInlineMapper _output_mapper;
private readonly BlueRowHasher _row_hasher;
private readonly TimeSpan _throttle_period;
private readonly ViewStreamDatabase _vsd;
private List<NonIncJoinBookkeeper> _book_keeping_data = new List<NonIncJoinBookkeeper>();
private IBlueTableSchema _output_schema = null;
private DateTime _last_fire_time = DateTime.MinValue;
private bool _is_dirty = false;
public NonIncJoinStream(ViewStream left, string left_join_cols, string left_proj,
ViewStream right, string right_join_cols, string right_proj)
: this(new Triplet<ViewStream, string, string>[]
{
new Triplet<ViewStream, string, string> (left, left_join_cols, left_proj),
new Triplet<ViewStream, string, string> (right, right_join_cols, right_proj)
}) { }
public NonIncJoinStream(params Triplet<ViewStream, string, string>[] join_specs)
: this(NonIncJoinSpec.from_triplet(join_specs), null, 0) { }
public NonIncJoinStream(ViewStreamDatabase vsd, INonIncJoinInlineMapper output_mapper, int throttles_ms,
params Triplet<ViewStream, string, string>[] join_specs)
: this(vsd, NonIncJoinSpec.from_triplet(join_specs), output_mapper, throttles_ms) {
}
public NonIncJoinStream(NonIncJoinSpec[] join_specs, INonIncJoinInlineMapper output_mapper, int throttles_ms = 0)
: this(ViewStreamDatabase.instance, join_specs, output_mapper, throttles_ms) { }
public NonIncJoinStream(ViewStreamDatabase vsd, NonIncJoinSpec[] join_specs, INonIncJoinInlineMapper output_mapper, int throttles_ms = 0) {
_join_specs = join_specs;
_output_mapper = output_mapper;
_row_hasher = new BlueRowHasher(_join_specs[0].join_col_names.Length);
_throttle_period = throttles_ms == 0 ? TimeSpan.Zero : new TimeSpan(0, 0, 0, 0, throttles_ms);
_vsd = vsd;
}
protected override void myOnFirstSubscriber() {
int data_col_no = 1;
for (int jp = 0; jp < _join_specs.Length; jp++) {
NonIncJoinBookkeeper s = new NonIncJoinBookkeeper(_join_specs[jp], jp, data_col_no, _row_hasher);
//Log.Gen.info($"NonIncJoinStream.myOnFirstSubscriber jp: {jp} spec: {_join_specs[jp]}");
data_col_no += _join_specs[jp].data_col_names.Length;
s.on_new_data += _stream_callback;
_book_keeping_data.Add(s);
}
if (_output_mapper != null)
_output_mapper.onMapperChange += _on_mapper_change;
_vsd.addWorker(this);
foreach (var bkd in _book_keeping_data)
bkd.init();
}
protected override void myOnLastSubscriber() {
foreach (NonIncJoinBookkeeper d in _book_keeping_data)
d.Dispose();
if (_output_mapper != null)
_output_mapper.onMapperChange -= _on_mapper_change;
_vsd.removeWorker(this);
}
private void _stream_callback() {
//Log.Gen.info("NonIncJoinStream._stream_callback received callback");
_is_dirty = true;
doWork();
}
private void _on_mapper_change() {
_is_dirty = true;
doWork();
}
private bool _is_ready() {
bool rv = true;
foreach (NonIncJoinBookkeeper d in _book_keeping_data)
rv &= d.is_ready;
return rv;
}
private void _construct_schema() {
if (_output_schema == null) {
List<ColumnInfo> cis = new List<ColumnInfo>(new ColumnInfo[] { new ColumnInfo("id", 0, typeof(int)) });
foreach (NonIncJoinBookkeeper d in _book_keeping_data)
d.constribute_cols_to_schema(cis);
_output_schema = new BlueTableSchema("NonIncJoinStream", cis.ToArray());
}
}
private int _n_total_fires = 0, _n_expensive_fires = 0;
private void _form_and_fire() {
//Log.Gen.info($"NonIncJoinStream._form_and_fire _is_ready(): {_is_ready()}");
if (_is_ready()) {
_construct_schema();
var timer = Stopwatch.StartNew();
IList<Pair<int, IBlueRow>> left_row_list = _book_keeping_data[0].get_list();
List<object[]> output_rows = new List<object[]>();
for (int i = 0; i < left_row_list.Count; i++) {
object[] or = new object[_output_schema.columnInfo.Length];
or[0] = i;
output_rows.Add(or);
}
for (int ji = 0; ji < _book_keeping_data.Count; ji++) {
NonIncJoinBookkeeper bk_data = _book_keeping_data[ji];
for (int i = 0; i < left_row_list.Count; i++) {
if (ji == 0)
bk_data.copy_master_row_to_output_row(i, output_rows[i]);
else
bk_data.copy_matching_row_to_output_row(left_row_list[i].first, output_rows[i]);
}
}
IBlueRowCollection dp_output_rows = _apply_map(new SequentialIDBlueTable(_output_schema, output_rows));
timer.Stop();
_n_total_fires++;
//if (timer.ElapsedMilliseconds >= 250) {
_n_expensive_fires++;
//string log_msg = $"Join and map time for {_book_keeping_data.Count} "
// + $"tables {timer.Elapsed.TotalMilliseconds} msecs, ratio {_n_expensive_fires}/{_n_total_fires}";
//Log.Gen.info(log_msg);
//}
var dp_out = BlueDataUtility.build_dp_inc(dp_output_rows, myLastValue as DataPacket, true);
myFireEvent(dp_out);
//myFireEvent(new DataPacket(dp_output_rows, dp_output_rows, myLastValue == null ? new BlueTable(_output_schema) : (myLastValue as DataPacket).newRows));
}
}
private IBlueRowCollection _apply_map(IBlueRowCollection unmapped_rows) {
if (_output_mapper != null) {
IBlueRowCollection output_rows;
var timer = Stopwatch.StartNew();
output_rows = _output_mapper.map_rows(unmapped_rows);
timer.Stop();
if (timer.ElapsedMilliseconds >= 50) {
Log.Gen.info($"Time to map {unmapped_rows.Count} rows into {output_rows.Count} rows {timer.Elapsed}");
}
return output_rows;
} else {
return unmapped_rows;
}
}
public void doWork() {
//Log.Gen.info($"NonIncJoinStream.doWork _last_fire_time: {_last_fire_time} _is_dirty: {_is_dirty}");
if (DateTime.Now - _last_fire_time >= _throttle_period && _is_dirty) {
_is_dirty = false;
_last_fire_time = DateTime.Now;
_form_and_fire();
}
}
}
}
@@ -0,0 +1,28 @@
//namespace bts.Blueorb {
// using BCG;
// public class JsonSaverStream : ViewStream {
// private ViewStream _stream;
// private string _file_name;
// public JsonSaverStream(ViewStream vs, string file_name) {
// _stream = vs;
// _file_name = file_name;
// }
// protected override void myOnFirstSubscriber() {
// _stream.onChange += _on_data;
// }
// private void _on_data(object dp_obj) {
// if (dp_obj == null) {
// return;
// }
// DataPacket dp = dp_obj as DataPacket;
// dp.newRows.write_to_json_file(dp.newRows, _file_name);
// }
// protected override void myOnLastSubscriber() {
// _stream.onChange -= _on_data;
// }
// }
//}
@@ -0,0 +1,91 @@
using System;
using System.Collections;
using System.Collections.Generic;
using Amazon.Runtime.Internal.Util;
using BCG;
using BCG.Utilities;
using bts;
namespace bts.BlueOrb{
public class LiteralStream : ViewStream {
public static List<Tuple<ColumnInfo, object>> parse_columns(string key_value_pairs) {
List<Tuple<ColumnInfo, object>> rv = new List<Tuple<ColumnInfo, object>>();
string[] pairs = key_value_pairs.Split(' ');
foreach (string p in pairs) {
string[] kvp = p.Split(':');
string col_name = kvp[0];
string val = kvp[1];
var tup = BlueDataUtility.guess_type(val);
var ci = new ColumnInfo(col_name, rv.Count, tup.Item1);
rv.Add(Tuple.Create(ci, tup.Item2));
}
return rv;
}
//literal("table1", "x:123, y:1.001, z:abc;x:123, y:1.001, z:abc")
// col_name: col_value ... ; col_name: col_value
public LiteralStream(string table_name, string rows) {
Func<string, int, IBlueRow> get_row = (json_row, id) => {
Dictionary<string, object> map = new Dictionary<string, object>();
string[] toks = json_row.Split(',', StringSplitOptions.TrimEntries);
//Logger.assert(col_vals.Length == 2);
foreach (string t in toks) {
var col_vals = t.Split(':', StringSplitOptions.TrimEntries);
string col = col_vals[0];
string val = col_vals[1];
map.Add(col, val);
}
map["id"] = id;
return new BlueRow(map);
};
string[] row_arr = rows.Split(';', StringSplitOptions.TrimEntries);
BlueTable table = null;
foreach (string json_string in row_arr) {
IBlueRow r = //BlueDataUtility.convert_json_to_row(json_string, table == null ? 0 : table.Count);
get_row(json_string, table == null ? 0 : table.Count);
table = table ?? new BlueTable(r.schema);
table.addRow(r);
}
ViewStream str = new BlueTableStream(table);
str = new SchemaGuesserStream(str);
mySetRealSource(str);
}
public LiteralStream(string tableName, string schema, string value) {
string[] columns = schema.Split(' ');
ColumnInfo[] ci = new ColumnInfo[columns.Length + 1];
ci[0] = new ColumnInfo("id", 0, typeof(int));
int n = 1;
foreach (string c in columns) {
string[] columnAndType = c.Split('/');
if (columnAndType.Length == 1)
ci[n] = new ColumnInfo(c, n, typeof(double));
else
ci[n] = new ColumnInfo(columnAndType[0], n, Type.GetType("System." + columnAndType[1], true));
n++;
}
BlueTable t = new BlueTable(new BlueTableSchema(tableName, ci));
string[] rows = value.Split(';');
n = 0;
foreach (string r in rows)
try {
if (r == string.Empty)
continue;
t.addRow(
BlueDataUtility.createRowFromCSVFileRow(r, t.schema, ++n));
} catch (Exception e) {
throw (new Exception("Error occured at line: " + n.ToString() + " " + e.Message));
}
mySetRealSource(new BlueRowCollectionStaticStream(t));
}
}
}
@@ -0,0 +1,89 @@
using System;
using BCG.Utilities.General;
using BCG;
using bts;
namespace bts.BlueOrb {
[DataStreamUserRespresentation("MillisecondStream")]
public class MillisecondStream : ViewStream, IWorker {
private DateTime _last_fire_time = DateTime.MinValue;
private readonly TimeSpan _period;
private readonly int _size;
private readonly ViewStreamDatabase _vsd;
private IBlueTableSchema _schema;
private bts.CircularBufferTable _table;
public MillisecondStream(int interval_msecs, int size)
: this(ViewStreamDatabase.instance, interval_msecs, size) {
}
public MillisecondStream(ViewStreamDatabase vsd, int interval_msecs, int size) {
_vsd = vsd;
_period = TimeSpan.FromMilliseconds(interval_msecs);
_size = size;
_schema = BlueDataUtility.form_schema("Milliseconds",
Tuple.Create("id", TypeCode.Int32),
Tuple.Create("time", TypeCode.DateTime),
Tuple.Create("display_time", TypeCode.String));
_table = new CircularBufferTable(_schema, _size);
}
protected override void myOnFirstSubscriber() {
_vsd.addWorker(this);
}
protected override void myOnLastSubscriber() {
_vsd.removeWorker(this);
}
private int _id = 0;
public void doWork() {
if (DateTime.Now - _last_fire_time >= _period) {
_last_fire_time = DateTime.Now;
IBlueRow nr = new BlueRow(_schema, _id++, _last_fire_time,
_last_fire_time.ToString("yyyy-MM-dd HH:mm:ss.fff"));
_table.add_row(nr);
var dp = BlueDataUtility.build_dp_inc(_table, myLastValue as DataPacket);
myFireEvent(dp);
}
}
}
public class MSecondStream : TimeUnitStream {
public MSecondStream(int intervalInms, int size)
: base(DateTime.Now, new TimeSpan(0, 0, 0, 0, intervalInms), size, ViewStreamDatabase.instance) { }
protected override DateTime myRoundTime(DateTime dt) { return dt; }
}
public class DisplayableMillisecondStream : ViewStream, IStreamMapper {
private IBlueTableSchema mySchema = new BlueTableSchema("ms",
new ColumnInfo("id", 0, typeof(int)),
new ColumnInfo("time", 1, typeof(string)));
public DisplayableMillisecondStream(int intervalInMs, int sz) {
mySetRealSource(new MapStream(new MSecondStream(intervalInMs, sz), this));
}
#region IStreamMapper Members
public IBlueTableSchema getOutputSchema(IBlueTableSchema inputSchema) {
return mySchema;
}
#endregion
#region IDynamicStreamMapper Members
public IBlueRow map(IBlueRow row) {
DateTime time = row.get_datetime("time");
var time_txt = time.ToString("yyyy-MM-dd HH:mm:ss.fff");
return new BlueRow(mySchema, row["id"], time_txt);
}
#endregion
private string myToString(DateTime dt) {
string sep = ":";
return dt.Hour + sep + dt.Minute + sep + dt.Second + sep + dt.Millisecond;
}
}
}
@@ -0,0 +1,127 @@
namespace bts.BlueOrb {
using BCG;
using System;
using System.Collections.Generic;
using System.Linq;
//using MathNet.Numerics.Statistics;
public enum MultiRowOp {
Mean,
Median,
Mode,
Range,
StdDev,
Max,
Min,
First,
Last,
Sum,
EdgeDiff,
EdgeDiffPercentage,
//IsEdgeDiff,
}
public class MultiRowOpHelper {
public readonly List<Tuple<string, MultiRowOp>> operations;
public static MultiRowOpHelper parse_ops(string text) {
// text is inp_col: op, inp_col: op, inp_col:op
List<Tuple<string, MultiRowOp>> rv = new List<Tuple<string, MultiRowOp>>();
string[] calcs = text.Split(',');
foreach (string c in calcs) {
string[] toks = c.Split(':');
string inp_col = toks[0].Trim();
string op_txt = toks[1].Trim();
MultiRowOp op = Enum.Parse<MultiRowOp>(op_txt, true);
rv.Add(Tuple.Create(inp_col, op));
}
return new MultiRowOpHelper(rv);
}
public MultiRowOpHelper(List<Tuple<string, MultiRowOp>> operations_) {
operations = operations_;
}
// aggregates multiple rows into one - ie does group_by compute
public static IBlueRow group_by_compute(List<IBlueRow> rows, MultiRowOpHelper parameters) {
if (rows.Count() == 0) {
return null;
}
IBlueRow ir = rows.ElementAt(0);
IBlueTableSchema in_schema = ir.schema;
IBlueTableSchema out_schema = in_schema;
IDictionary<string, Tuple<string, MultiRowOp>> out_col_map =
new Dictionary<string, Tuple<string, MultiRowOp>>();
foreach (var t in parameters.operations) {
string out_col_name = $"{t.Item1}_{t.Item2}".ToLower();
out_schema = out_schema.add_column(TypeCode.Double, out_col_name);
out_col_map[out_col_name] = t;
}
MutableBlueRow or = new MutableBlueRow(out_schema, ir);
foreach (var kv in out_col_map) {
string out_col_name = kv.Key;
string in_col_name = kv.Value.Item1;
MultiRowOp op = kv.Value.Item2;
IEnumerable<double> col_vals = BlueDataUtility.get_colvalues<double>(rows, in_col_name);
double comp_res = compute_op(col_vals, op);
or.setValue(out_col_name, comp_res);
}
return or;
}
// rolling_rows are the rolling window of rows. the 0th row is the one thats gettin
// calculated. rows 1 - n-1 are its window rows. the schema is based on the operations.
public IBlueRow rolling_calc(List<IBlueRow> rolling_rows) {
if (rolling_rows.Count() == 0) {
return null;
}
IBlueRow ir = rolling_rows.ElementAt(0);
IBlueTableSchema in_schema = ir.schema;
IBlueTableSchema out_schema = in_schema;
IDictionary<string, Tuple<string, MultiRowOp>> out_col_map =
new Dictionary<string, Tuple<string, MultiRowOp>>();
foreach (var t in operations) {
string out_col_name = $"{t.Item1}_{t.Item2}".ToLower();
out_schema = out_schema.add_column(TypeCode.Double, out_col_name);
out_col_map[out_col_name] = t;
}
MutableBlueRow or = new MutableBlueRow(out_schema, ir);
foreach (var kv in out_col_map) {
string out_col_name = kv.Key;
string in_col_name = kv.Value.Item1;
MultiRowOp op = kv.Value.Item2;
IEnumerable<double> col_vals = BlueDataUtility.get_colvalues<double>(rolling_rows, in_col_name);
double comp_res = compute_op(col_vals, op);
or.setValue(out_col_name, comp_res);
}
return or;
}
public static double compute_op(IEnumerable<double> col_vals, MultiRowOp op) {
switch (op) {
case MultiRowOp.Max:
return col_vals.Max<double>();
case MultiRowOp.Min:
return col_vals.Min<double>();
case MultiRowOp.Mean:
return col_vals.Sum() / col_vals.Count();
case MultiRowOp.Sum:
return col_vals.Sum();
case MultiRowOp.First:
return col_vals.First();
case MultiRowOp.Last:
return col_vals.Last();
case MultiRowOp.StdDev:
return MathNet.Numerics.Statistics.Statistics.StandardDeviation(col_vals);
default:
// TODO
Log.assert(false);
return Double.NaN;
}
}
}
}
@@ -0,0 +1,207 @@
namespace bts.BlueOrb {
using BCG;
using ProtoBuf;
using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
#region mappings
internal class ProjectData {
// schema from the base stream
public readonly IBlueTableSchema in_schema;
// this is the projected schema
public readonly IBlueTableSchema out_schema;
// projected cols to the input cols, for cols who's names are changed
public readonly IDictionary<string, string> mapped_to_orig_col_map = new Dictionary<string, string>();
public ProjectData(IBlueTableSchema in_schema_, IBlueTableSchema out_schema_,
IDictionary<string, string> mapped_to_orig_col_map_) {
in_schema = in_schema_;
out_schema = out_schema_;
mapped_to_orig_col_map = mapped_to_orig_col_map_;
}
}
#endregion
#region Row with mapipings build in
internal class NewProjectRow : IBlueRow {
#region IBlueRow members
private ProjectData _project_data;
private IBlueRow _row;
public NewProjectRow(ProjectData project_data, IBlueRow row) {
_project_data = project_data;
_row = row;
}
public IBlueTableSchema schema { get { return _project_data.out_schema; } }
#endregion
#region ICollection members
public int Count { get { return schema.columnNames.Length; } }
public bool IsSynchronized { get { return false; } }
public object SyncRoot { get { return this; } }
public void CopyTo(Array array, int index) {
int i = index;
foreach (object o in this as IEnumerable)
array.SetValue(o, i++);
}
#endregion
#region IDictionary members
public bool IsFixedSize { get { return true; } }
public bool IsReadOnly { get { return true; } }
public object this[object key] {
get {
if (key is string)
return this[(string)key];
return this[(int)key];
}
set { throw new NotSupportedException("The IBlueRow is readonly"); }
}
public object this[int index] {
get {
return this[schema.columnInfo[index].colName];
}
}
public object this[string col_name] {
get {
string mapped_to_col = null;
if (_project_data.mapped_to_orig_col_map.TryGetValue(col_name, out mapped_to_col)) {
return _row[mapped_to_col];
}
return _row[col_name];
}
}
public ICollection Keys { get { return schema.columnNames; } }
public ICollection Values {
get {
object[] retVal = new object[schema.columnNames.Length];
CopyTo(retVal, 0);
return retVal;
}
}
public void Add(object key, object value) { throw new NotSupportedException("The IBlueRow is readonly"); }
public void Clear() { throw new NotSupportedException("The IBlueRow is readonly"); }
public bool Contains(object key) { return schema[(string)key] != null; }
/// <summary>
/// Note, the default enumerator is the IDictionary's enumerator, so if user needs to
/// iterate through the values, typecast this into an IEnumerable
/// </summary>
/// <returns></returns>
public IDictionaryEnumerator GetEnumerator() { return new BlueRowDictionaryEnumerator(this); }
IEnumerator IEnumerable.GetEnumerator() { return new BlueRowEnumerator(this); }
public void Remove(object key) { throw new NotSupportedException("The IBlueRow is readonly"); }
#endregion
}
#endregion
[DataStreamUserRespresentation("ProjectStream")]
public class NewProjectStream : ViewStream, IStreamMapper {
private ViewStream _stream;
private string _project_expression;
private ProjectData _project_data = null;
//public ProjectData
public NewProjectStream(ViewStream vs, string project_expression) {
_stream = vs;
_project_expression = project_expression;
_stream = new MapStream(vs, this);
mySetRealSource(_stream);
}
public IBlueTableSchema getOutputSchema(IBlueTableSchema input_schema) {
_project_data = _create_project_data(input_schema, _project_expression);
return _project_data.out_schema;
}
public IBlueRow map(IBlueRow row) {
return new NewProjectRow(_project_data, row);
}
private static ProjectData _create_project_data(IBlueTableSchema input_schema,
string project_expression) {
project_expression = project_expression.Trim();
string[] project_toks = project_expression.Split(' ');
HashSet<string> removed_cols = new HashSet<string>();
Dictionary<string, string> mapped_to_orig_map = new Dictionary<string, string>();
Dictionary<string, string> orig_to_mapped = new Dictionary<string, string>();
// by default every col will show up, so if we have the projection as (a:a1)
// it will convert a to a1,
// examples: if a schema has cols a b c
// project(parent_str, "a:aa ") -> aa b c
// project(parent_str, "a:aa !c") -> aa b
// project(parent_str, "a:aa !") -> aa
// project(parent_str, "a:aa b")) -> aa b
bool hide_all = false;
foreach (string p_tok in project_toks) {
if (p_tok.StartsWith("!")) {
if (p_tok.Length == 1) {
// if it is a "blanket "!"=> hide all cols except the one that
// are stated explicitly
hide_all = true;
} else {
string removed_col = p_tok.Substring(1);
removed_cols.Add(removed_col);
}
} else if (p_tok.Contains(":")) {
string[] map_toks = p_tok.Split(":");
string orig_col = map_toks[0];
string mapped_col = map_toks[1];
mapped_to_orig_map.Add(mapped_col, orig_col);
orig_to_mapped.Add(orig_col, mapped_col);
} else {
if (input_schema.columnNames.Contains(p_tok)) {
string orig_col = p_tok;
string mapped_col = orig_col;
orig_to_mapped.Add(orig_col, mapped_col);
}
}
}
// make the new schema
List<ColumnInfo> new_cols = new List<ColumnInfo>();
foreach (ColumnInfo ci in input_schema.columnInfo) {
if (ci.colName == "id") {
ColumnInfo nc = new ColumnInfo(ci.colName, new_cols.Count, ci.colType);
new_cols.Add(nc);
continue;
}
if (removed_cols.Contains(ci.colName)) {
continue;
} else if (orig_to_mapped.ContainsKey(ci.colName)) {
string mapped_name = orig_to_mapped[ci.colName];
ColumnInfo nc = new ColumnInfo(mapped_name, new_cols.Count, ci.colType);
new_cols.Add(nc);
} else {
if (hide_all)
continue;
//if (ci.colName == "id" || removed_cols.Count > 0) {
// unless columns are explicitely hidden, all columns show up
ColumnInfo nc = new ColumnInfo(ci.colName, new_cols.Count, ci.colType);
new_cols.Add(nc);
//}
}
}
IBlueTableSchema new_schema = new BlueTableSchema(input_schema.tableName, new_cols.ToArray());
return new ProjectData(input_schema, new_schema, mapped_to_orig_map);
}
}
}
@@ -0,0 +1,30 @@
using BCG;
namespace bts.BlueOrb {
public class SchemaGuesserStream : ViewStream {
private IBlueRowCollection _rows = null;
private ViewStream _vs;
public SchemaGuesserStream(ViewStream vs) {
_vs = vs;
}
protected override void myOnFirstSubscriber() {
_vs.onChange += _on_data;
}
protected override void myOnLastSubscriber() {
_vs.onChange -= _on_data;
}
private void _on_data(object obj) {
DataPacket dp = obj as DataPacket;
if (_rows == null) {
_rows = BlueDataUtility.get_typed_table(dp.newRows);
}
var dp_out = BlueDataUtility.build_dp_inc(_rows, myLastValue as DataPacket);
myFireEvent(dp_out);
}
}
}
@@ -0,0 +1,67 @@
using Bts.ArithmeticExpressions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using BCG;
namespace bts.BlueOrb {
[DataStreamUserRespresentation("SingleRowMathStream")]
public class MathStream : ViewStream, IStreamMapper {
private IArithmeticProgram myExp;
private BooleanExpressionFilter _bool;
List<string> myExtraColumns = new List<string>();
BlueTableSchema mySchema;
public MathStream(ViewStream vs, string mathExpression)
: this(vs, mathExpression, string.Empty) {
}
public MathStream(ViewStream vs, string mathExpression, string boolean_expression) {
myExp = ArithmeticProgramParser.Parse(mathExpression) ??
throw new Exception($"Failed to parse ArithmeticExpression '{mathExpression}'");
myExtraColumns = myExp.OutputIdentifiers.ToList();
if (!string.IsNullOrEmpty(boolean_expression))
_bool = new BooleanExpressionFilter(boolean_expression);
mySetRealSource(new MapStream(vs, this));
}
#region IStreamMapper Members
public IBlueTableSchema getOutputSchema(IBlueTableSchema inputSchema) {
if (mySchema == null) {
Dictionary<string, ColumnInfo> colIdx = new Dictionary<string, ColumnInfo>();
foreach (ColumnInfo ci in inputSchema.columnInfo)
colIdx[ci.colName] = ci;
foreach (string ncn in myExtraColumns) {
ColumnInfo newC;
if (colIdx.ContainsKey(ncn))
newC = new ColumnInfo(colIdx[ncn].colName, colIdx[ncn].colNo, typeof(double));
else
newC = new ColumnInfo(ncn, colIdx.Count, typeof(double));
colIdx[ncn] = newC;
}
mySchema = new BlueTableSchema(colIdx.Values.ToArray(), "math(" + inputSchema.tableName + ")");
}
return mySchema;
}
#endregion
#region IDynamicStreamMapper Members
public IBlueRow map(IBlueRow row) {
MutableBlueRow result = new MutableBlueRow(mySchema, row);
if (_bool == null || _bool.filter(row)) {
myExp.Evaluate(result.AsDictionary());
}
return result;
}
#endregion
}
}
@@ -0,0 +1,11 @@
namespace bts.BlueOrb {
using BCG;
[DataStreamUserRespresentation("TopStream")]
public class TopStream : ViewStream {
public TopStream(ViewStream vs, int n_rows, string sort_col) {
ViewStream rs = new BCG.TopStream (vs, n_rows, sort_col);
mySetRealSource(rs);
}
}
}
@@ -0,0 +1,261 @@
using System;
using System.Collections.Generic;
using mbts;
using BCG;
namespace bts.BlueOrb {
internal class MutableAppendTable : IBlueRowCollection {
private class MyRow : IBlueRow {
private IBlueRow myRow;
private int myID;
public MyRow(IBlueRow r, int id) {
myRow = r;
myID = id;
}
#region IBlueRow Members
public IBlueTableSchema schema { get { return myRow.schema; } }
public object this[int index] {
get {
if (myRow.schema[index].colName == "id")
return myID;
else
return myRow[index];
}
}
public object this[string colName] { get { return colName == "id" ? (object)myID : myRow[colName]; } }
#endregion
#region IDictionary Members
public void Add(object key, object value) { throw new Exception("Row is read only."); }
public void Clear() { throw new Exception("Row is read only."); }
public bool Contains(object key) { return myRow.Contains(key); }
public System.Collections.IDictionaryEnumerator GetEnumerator() { return new BlueRowDictionaryEnumerator(this); }
public bool IsFixedSize { get { return true; } }
public bool IsReadOnly { get { return true; } }
public System.Collections.ICollection Keys { get { return myRow.Keys; } }
public void Remove(object key) { throw new Exception("The method or operation is not implemented."); }
public System.Collections.ICollection Values {
get {
object[] vals = new object[myRow.Count];
CopyTo(vals, 0);
return vals;
}
}
public object this[object key] {
get { return key is string ? this[key as string] : this[(int)key]; }
set { throw new Exception("The method or operation is not implemented."); }
}
#endregion
#region ICollection Members
public void CopyTo(Array array, int index) {
myRow.CopyTo(array, index);
array.SetValue(myID, index + myRow.schema["id"].colNo);
}
public int Count { get { return myRow.Count; } }
public bool IsSynchronized { get { return myRow.IsSynchronized; } }
public object SyncRoot { get { return myRow.SyncRoot; } }
#endregion
#region IEnumerable Members
System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() { return new BlueRowEnumerator(this); }
#endregion
}
private BlueTable myRows;
private int myID = 0;
/// <summary>
/// maps the table no to a dictionary containing the
/// local IDs to a row in the final big table
/// </summary>
private Dictionary<int, Dictionary<int, IBlueRow>> myIDMap = new Dictionary<int, Dictionary<int, IBlueRow>>();
public MutableAppendTable(IBlueTableSchema schema) { myRows = new BlueTable(schema); }
public IBlueRow addRow(IBlueRow r, int tableNo) {
if (!myIDMap.ContainsKey(tableNo))
myIDMap.Add(tableNo, new Dictionary<int, IBlueRow>());
IBlueRow nr = new MyRow(r, myID++); // BlueDataUtility.getNewReplacedRow(r, myID++, "id");
myRows.addRow(nr);
myIDMap[tableNo].Add((int)r["id"], nr);
return nr;
}
public IBlueRow removeRow(IBlueRow r, int tableNo) {
IBlueRow dr = myIDMap[tableNo][(int)r["id"]];
myRows.deleteRow(dr);
myIDMap[tableNo].Remove((int)r["id"]);
return dr;
}
public IBlueRowCollection removeTable(int tableNo) {
if (!myIDMap.ContainsKey(tableNo))
return null;
else {
BlueTable rv = new BlueTable(myRows.schema);
foreach (IBlueRow r in myIDMap[tableNo].Values)
rv.addRow(r);
myRows.deleteRows(rv);
myIDMap.Remove(tableNo);
return rv;
}
}
public void clear() {
myRows.clear();
myIDMap.Clear();
}
#region IBlueRowCollection Members
public IBlueRow this[int id] { get { return myRows[id]; } }
public IBlueTableSchema schema { get { return myRows.schema; } }
public bool containsID(int id) { return myRows.containsID(id); }
#endregion
#region ICollection Members
public void CopyTo(Array array, int index) { myRows.CopyTo(array, index); }
public int Count { get { return myRows.Count; } }
public bool IsSynchronized { get { return myRows.IsSynchronized; } }
public object SyncRoot { get { return myRows.SyncRoot; } }
#endregion
#region IEnumerable Members
public System.Collections.IEnumerator GetEnumerator() { return myRows.GetEnumerator(); }
#endregion
}
/// <summary>
/// MutableStream can append multiple streams
/// as well as give the user the flexibility
/// to append (or remove) more streams at
/// runtime
/// </summary>
}
namespace bts.BlueOrb {
[DataStreamUserRespresentation("UnionStream")]
public class MutableAppendStream : ViewStream {
private bool mySubscriptionActive = false;
private MutableAppendTable myTable;
private int myMovingIndex;
private Dictionary<int, Pair<ViewStream, ViewStream>> myStreamIndex = new Dictionary<int, Pair<ViewStream, ViewStream>>();
private bool _allign_col_orders = true;
public MutableAppendStream(params ViewStream[] streams)
: this(new List<ViewStream>(streams)) { }
public MutableAppendStream(bool allign_col_order, params ViewStream[] streams)
: this(allign_col_order, new List<ViewStream>(streams)) { }
public MutableAppendStream(ICollection<ViewStream> streams)
: this(true, streams) { }
public MutableAppendStream(bool allign_col_orders, ICollection<ViewStream> streams) {
_allign_col_orders = allign_col_orders;
foreach (ViewStream vs in streams) {
myMovingIndex++;
ViewStream pj = new ChangeTableNameStream(vs, myMovingIndex.ToString());
if (_allign_col_orders)
pj = new ProjectStream(pj, "!");
myStreamIndex.Add(myMovingIndex, new Pair<ViewStream, ViewStream>(vs, pj));
}
}
public void addStream(ViewStream vs) {
myMovingIndex++;
///the change name is done to identify the callbacks, and the project to
///make all the schemas col orders the same
ViewStream pj = new ChangeTableNameStream(vs, myMovingIndex.ToString());
if (_allign_col_orders)
pj = new ProjectStream(pj, "!");
myStreamIndex.Add(myMovingIndex, new Pair<ViewStream, ViewStream>(vs, pj));
if (mySubscriptionActive)
pj.onChange += new ChangeDelegate(myHandleStreamChanges);
}
public void removeStream(ViewStream vs) {
int index = -1;
foreach (KeyValuePair<int, Pair<ViewStream, ViewStream>> p in myStreamIndex) {
if (p.Value.first == vs) {
index = p.Key;
if (mySubscriptionActive) {
p.Value.second.onChange -= new ChangeDelegate(myHandleStreamChanges);
if (myTable != null) {
IBlueRowCollection delRows = myTable.removeTable(p.Key);
if (delRows != null)
myFireEvent(new DataPacket(myTable, new BlueTable(myTable.schema), delRows));
}
}
break;
}
if (index > 0)
myStreamIndex.Remove(index);
}
}
protected override void myOnFirstSubscriber() {
mySubscriptionActive = true;
foreach (KeyValuePair<int, Pair<ViewStream, ViewStream>> p in myStreamIndex)
p.Value.second.onChange += new ChangeDelegate(myHandleStreamChanges);
}
protected override void myOnLastSubscriber() {
mySubscriptionActive = false;
foreach (KeyValuePair<int, Pair<ViewStream, ViewStream>> p in myStreamIndex)
p.Value.second.onChange -= new ChangeDelegate(myHandleStreamChanges);
if (myTable != null)
myTable.clear();
}
private void myHandleStreamChanges(object newValue) {
DataPacket dp = newValue as DataPacket;
myTable = myTable ?? new MutableAppendTable(new BlueTableSchema("MutableAppend", dp.schema.columnInfo));
BlueTable adds = new BlueTable(myTable.schema),
dels = new BlueTable(myTable.schema);
int index = Int32.Parse(dp.schema.tableName);
foreach (IBlueRow r in dp.deletedRows) {
IBlueRow dr = myTable.removeRow(r, index);
dels.addRow(dr);
}
foreach (IBlueRow r in dp.addedRows) {
IBlueRow nr = myTable.addRow(r, index);
adds.addRow(nr);
}
myFireEvent(new DataPacket(myTable, adds, dels));
}
}
}
@@ -18,6 +18,7 @@
<ItemGroup>
<PackageReference Include="HtmlAgilityPack" Version="1.11.57" />
<PackageReference Include="IronPython" Version="3.4.2" />
<PackageReference Include="Microsoft.Data.Analysis" Version="0.21.1" />
<PackageReference Include="MongoDB.Bson" Version="2.23.1" />
<PackageReference Include="MongoDB.Driver" Version="2.23.1" />
@@ -1,52 +1,47 @@
using System;
namespace BCG
{
[DataStreamUserRespresentation ("count")]
public class CountStream : ViewStream
{
namespace bts.BlueOrb {
using BCG;
[DataStreamUserRespresentation("CountStream")]
public class CountStream : ViewStream {
private DataPacket myLastFiredDataPacket;
private IBlueTableSchema mySchema = new BlueTableSchema ("count",
new ColumnInfo ("id", 0, typeof(int)),
new ColumnInfo ("count", 1, typeof(int)));
private IBlueTableSchema mySchema = new BlueTableSchema("count",
new ColumnInfo("id", 0, typeof(int)),
new ColumnInfo("count", 1, typeof(int)));
private int myId = 0;
private ViewStream myVS;
public CountStream (ViewStream vs)
{
public CountStream(ViewStream vs) {
myVS = vs;
}
public CountStream (string tableName)
: this (new TableStream (tableName))
{}
public CountStream(string tableName)
: this(new TableStream(tableName)) { }
protected override void myOnFirstSubscriber () {
protected override void myOnFirstSubscriber() {
myVS.onChange += new ChangeDelegate(myHandleChanges);
}
protected override void myOnLastSubscriber () {
protected override void myOnLastSubscriber() {
myVS.onChange -= new ChangeDelegate(myHandleChanges);
}
private void myHandleChanges(object newValue)
{
myFireEvent (
private void myHandleChanges(object newValue) {
myFireEvent(
myLastFiredDataPacket =
myLastFiredDataPacket == null ?
new DataPacket (myGetRows (newValue, ++myId), myGetRows (newValue, myId),
new BlueTable (mySchema))
new DataPacket(myGetRows(newValue, ++myId), myGetRows(newValue, myId),
new BlueTable(mySchema))
:
new DataPacket (myGetRows (newValue, ++myId), myGetRows (newValue, myId),
new DataPacket(myGetRows(newValue, ++myId), myGetRows(newValue, myId),
myLastFiredDataPacket.newRows)
);
}
private IBlueRowCollection myGetRows (object dp, int id)
{
BlueTable retVal = new BlueTable (mySchema);
private IBlueRowCollection myGetRows(object dp, int id) {
BlueTable retVal = new BlueTable(mySchema);
retVal.addRow
(new BlueRow (mySchema, id, (dp as DataPacket).newRows.Count));
(new BlueRow(mySchema, id, (dp as DataPacket).newRows.Count));
return retVal;
}
@@ -5,25 +5,20 @@ using mbts.config;
using System;
using System.Text.RegularExpressions;
namespace BCG.StreamWorld.Addition
{
[DataStreamUserRespresentation (false)]
public class FilterStream : ViewStream
{
namespace BCG.StreamWorld.Addition {
[DataStreamUserRespresentation(false)]
public class FilterStream : ViewStream {
private ViewStream myBaseStream;
private BlueTable myBaseTable, myStateTable;
private IStreamFilter myFilter;
public FilterStream(ViewStream vs)
: this(vs, "1=1")
{ }
: this(vs, "1=1") { }
public FilterStream(ViewStream vs, string filterExp)
: this(vs, new BooleanExpressionFilter(filterExp))
{ }
: this(vs, new BooleanExpressionFilter(filterExp)) { }
public FilterStream(ViewStream vs, IStreamFilter filter)
{
public FilterStream(ViewStream vs, IStreamFilter filter) {
myBaseStream = vs;
myFilter = filter;
if (filter is IDynamicStreamFilter)
@@ -31,43 +26,35 @@ namespace BCG.StreamWorld.Addition
}
protected override void myOnFirstSubscriber()
{
protected override void myOnFirstSubscriber() {
myBaseStream.onChange += new ChangeDelegate(myHandleChanges);
}
protected override void myOnLastSubscriber()
{
protected override void myOnLastSubscriber() {
myBaseStream.onChange -= new ChangeDelegate(myHandleChanges);
myBaseTable.clear();
myStateTable.clear();
}
private void myHandleChanges(object newValue)
{
private void myHandleChanges(object newValue) {
DataPacket dp = newValue as DataPacket;
if (myStateTable == null)
{
if (myStateTable == null) {
bts.Log.assert(myBaseTable == null);
myStateTable = new BlueTable(dp.schema);
myBaseTable = new BlueTable(dp.schema);
}
BlueTable dels = new BlueTable(dp.schema),
adds = new BlueTable(dp.schema);
foreach (IBlueRow r in dp.deletedRows)
{
foreach (IBlueRow r in dp.deletedRows) {
myBaseTable.deleteRow(r);
if (myStateTable.containsID((int)r["id"]))
{
if (myStateTable.containsID((int)r["id"])) {
dels.addRow(r);
myStateTable.deleteRow(r);
}
}
foreach (IBlueRow r in dp.addedRows)
{
foreach (IBlueRow r in dp.addedRows) {
myBaseTable.addRow(r);
if (myFilter.filter(r))
{
if (myFilter.filter(r)) {
adds.addRow(r);
myStateTable.addRow(r);
}
@@ -76,13 +63,10 @@ namespace BCG.StreamWorld.Addition
myFireEvent(new DataPacket(myStateTable, adds, dels));
}
public IStreamFilter filter
{
set
{
public IStreamFilter filter {
set {
myFilter = value;
if (myBaseTable != null)
{
if (myBaseTable != null) {
IBlueRowCollection oldState = new BlueTable(myStateTable);
myStateTable.clear();
foreach (IBlueRow r in myBaseTable)
@@ -5,7 +5,7 @@ using System.Collections.Generic;
using System.Linq;
namespace BCG {
public class GroupByStream : MethodDrivenStream {
internal class GroupByStream : MethodDrivenStream {
private class MyFuncGroupByEvaluator : IGroupByEvaluator {
private string[] _groupby_cols;
private Func<DataPacket, IBlueRow> _groupby_func;
@@ -40,4 +40,4 @@ namespace BCG {
myFireSwitch = myLastFiredPacket.addedRows.Count + myLastFiredPacket.deletedRows.Count == 0 ? false : true;
}
}
}
}
@@ -280,6 +280,28 @@ namespace BCG {
#region json
public static void write_to_json_file(this IBlueRowCollection table, string file_path) {
write_dict_to_json_file(file_path, table.to_namevalue_pairs());
}
public static void write_dict_to_json_file(string file_path, List<Dictionary<string, string>> data) {
var options = new JsonSerializerOptions { WriteIndented = true };
string json = System.Text.Json.JsonSerializer.Serialize(data, options);
File.WriteAllText(file_path, json);
}
public static List<Dictionary<string, string>> ReadFromFile(string filePath) {
string json = File.ReadAllText(filePath);
return System.Text.Json.JsonSerializer.Deserialize<List<Dictionary<string, string>>>(json);
}
public static void write_table_to_json(IBlueRowCollection table) {
List<Dictionary<string, string>> rows = new List<Dictionary<string, string>>();
foreach (IBlueRow row in table) {
row.to_json();
}
}
public static Dictionary<string, string> convert_json_to_dictionary(string json_string) {
Dictionary<string, string> dictionary = JsonConvert.DeserializeObject<Dictionary<string, string>>(json_string);
return dictionary;
@@ -348,6 +370,14 @@ namespace BCG {
}
return rv;
}
public static List<Dictionary<string,string>> to_namevalue_pairs(this IBlueRowCollection table) {
List<Dictionary<string, string>> rv = new List<Dictionary<string, string>>();
foreach (IBlueRow r in table) {
rv.Add(r.to_dict());
}
return rv;
}
#region Schema types
// these are the only types allowed in schemas
private static IReadOnlyDictionary<TypeCode, Type> _typecode_to_type = new Dictionary<TypeCode, Type>
@@ -1225,6 +1255,20 @@ namespace BCG {
return rv;
}
public static Dictionary<string, string> to_dict(this IBlueRow r) {
Dictionary<string, string> map = new Dictionary<string, string>();
foreach (ColumnInfo c in r.schema.columnInfo) {
map.Add(c.colName, r[c.colName].ToString());
}
return map;
}
public static string to_json(this IBlueRow r) {
Dictionary<string, string> map = r.to_dict();
string json = System.Text.Json.JsonSerializer.Serialize(map);
return json;
}
public static void compute_differences_fast(IBlueRowCollection new_rows, IBlueRowCollection old_rows,
out IBlueRowCollection out_d_rows, out IBlueRowCollection out_a_rows) {
if (old_rows == null || old_rows.Count == 0) {
@@ -47,10 +47,10 @@ namespace BCG {
/// </summary>
/// <param name="assemblyName"></param>
public void loadAssembly(string assemblyName) {
Func<Type, bool> is_silverpine_only = (t) => {
if (!assemblyName.Contains("SharedStreams.dll"))
return true;
return t.Namespace == "bts.SilverPine";
Func<Type, bool> is_blue_orb = (t) => {
//if (!assemblyName.Contains("SharedStreams.dll"))
// return true;
return t.Namespace == "bts.BlueOrb";// || t.Namespace == "BCG";
};
bts.Log.Gen.info($"PrimitiveManager.loadAssembly assemblyName: {assemblyName}");
if (myAssemblyNames.Contains(assemblyName))
@@ -61,7 +61,7 @@ namespace BCG {
Assembly.LoadFrom(assemblyName);
foreach (Type type in loadedAssembly.GetTypes())
if (!type.IsInterface && is_silverpine_only(type)
if (!type.IsInterface && is_blue_orb(type)
&& !type.IsAbstract
&& type.IsSubclassOf(typeof(ViewStream))
&& type.IsPublic) {
@@ -77,7 +77,7 @@ namespace BCG {
break;
}
if (isVisible) {
Log.assert_d(myTypeLookupTable[userRep] == null, type.FullName);
// Log.assert_d(myTypeLookupTable[userRep] == null, type.FullName);
myTypeLookupTable[userRep] = type;
}
}
@@ -1,4 +1,5 @@
namespace bts.SilverPine {
namespace bts.BlueOrb {
using bts.SilverPine;
using BCG;
using System.Runtime.CompilerServices;
@@ -45,7 +46,7 @@ namespace bts.SilverPine {
}
[DataStreamUserRespresentation("FuturesHistorical1DayBars")]
public class FuturesHistorical1DayBarsStream : ViewStream {
internal class FuturesHistorical1DayBarsStream : ViewStream {
// this gets the raw bars for a symbol (perhaps make it into a raw bars stream?)
public FuturesHistorical1DayBarsStream(string symbol) :
@@ -103,7 +104,7 @@ namespace bts.SilverPine {
}
[DataStreamUserRespresentation("FuturesHistorical1MinBars")]
public class FuturesHistorical1MinuteBarsStream : ViewStream {
internal class FuturesHistorical1MinuteBarsStream : ViewStream {
// same as the above but for one min
// this gets the raw bars for a symbol
public FuturesHistorical1MinuteBarsStream(string symbol) :
@@ -187,46 +188,6 @@ namespace bts.SilverPine {
}
}
[DataStreamUserRespresentation("FuturePriceDiff")]
public class FutureHistoricalBarsPriceDiffStream : ViewStream {
public FutureHistoricalBarsPriceDiffStream(string future_contract, string dt1, string dt2)
:
this(new ContractUtils.FutureContract(future_contract), DateTime.Parse(dt1), DateTime.Parse(dt2), "close", ViewStreamDatabase.instance) {
}
public FutureHistoricalBarsPriceDiffStream(string future_contract, DateTime dt1, DateTime dt2)
:
this(new ContractUtils.FutureContract(future_contract), dt1, dt2, "close", ViewStreamDatabase.instance) {
}
public FutureHistoricalBarsPriceDiffStream(string future_contract, DateTime dt1, DateTime dt2, string col)
:
this(new ContractUtils.FutureContract(future_contract), dt1, dt2, col, ViewStreamDatabase.instance) {
}
public FutureHistoricalBarsPriceDiffStream(ContractUtils.FutureContract fc, DateTime dt1, DateTime dt2, string col, ViewStreamDatabase vsd) {
BarSize bsize = dt2 - dt1 >= TimeSpan.FromDays(1.0) ? BarSize.ONE_DAY : BarSize.ONE_MIN;
ViewStream rs = new FutureHistoricalBarsStream(fc, BarSize.ONE_DAY, new DateTimeRange(dt1, dt2));
rs = new ColumnDifferenceStream(rs, r => {
DateTime r_dt = r.get_datetime("ts_event");
var rv = dt1.Equals(r_dt);
return rv;
},
//r => dt2.Equals(r.get_datetime("ts_event")), col, col);
r => {
DateTime r_dt = r.get_datetime("ts_event");
var rv = dt2.Equals(r_dt);
return rv;
}, col, col);
rs = new NewProjectStream(rs, "diff !");
mySetRealSource(rs);
}
}
// takes the contract name and expands it into multuiple columns corresponding to the contracts.
// the contract columns are a union, of all properties with some of them empty
public class ContractExpanderStream : ViewStream {
@@ -247,3 +208,29 @@ namespace bts.SilverPine {
}
}
namespace bts.BlueOrb {
using BCG;
using bts.api;
using Newtonsoft.Json.Serialization;
[DataStreamUserRespresentation("FuturesContinousContract")]
public class FuturesContinousContractStream : ViewStream {
// the argument is symbol, like "ES" etc and not the contract itself
public FuturesContinousContractStream(string symbol) : this(symbol, ViewStreamDatabase.instance) { }
public FuturesContinousContractStream(string symbol, ViewStreamDatabase db) {
symbol = symbol.EndsWith("*") ? symbol : symbol + "*";
ViewStream r = new FuturesHistorical1DayBarsStream(symbol);
r = new ContractExpanderStream(r, "symbol");
r = new FilterStream(r, "contract_type='Future'");
// finding the max volume across symbols, if we need to do this for open_interest, then join here
// with the open interest stream on symbol/contract
r = new NonIncGroupByStream(r, "ts_event", "volume:Max");
r = new NewProjectStream(r, "ts_event symbol !");
r = new SortStream(r, "-ts_event");
mySetRealSource(r);
}
}
}
@@ -1,4 +1,5 @@
namespace bts.SilverPine {
namespace bts.BlueOrb {
using bts.SilverPine;
using BCG;
using System;
@@ -1,117 +1,118 @@
// namespace bts.SilverPine {
// using bts;
// using BCG;
// using MongoDB.Driver.Linq;
// using System.ComponentModel.DataAnnotations;
// using ZstdSharp.Unsafe;
// TODO: Remove
//// namespace bts.SilverPine
//// using bts;
//// using BCG;
//// using MongoDB.Driver.Linq;
//// using System.ComponentModel.DataAnnotations;
//// using ZstdSharp.Unsafe;
// [DataStreamUserRespresentation("VWAP")]
// public class VWAPStream : ViewStream {
// // returns the VWAP stream for each contract
// private static ViewStream build_vwap_stream(string symbol, DateTime begin_dt, DateTime end_dt) {
// var ohlcv_params = OHLCVStream.get_params(begin_dt, end_dt);
// ViewStream ohlc_str = new OHLCVStream(symbol, ohlcv_params);
// // stream is grouped per contract and then vwap is calculated per contract
// GlobalMapStream rs = new GlobalMapStream(ohlc_str, "ts_event", "contract",
// (rows, idx) => {
// Log.Gen.info($"VWAPStream.GlobalMapStream delegate n_rows: {rows.Count} idx: {idx}");
// IBlueRow ir = rows.ElementAt((int)idx);
// var sel_rows = rows.Skip((int)idx).Where(r => r.get_datetime("ts_event") >= end_dt && r.get_datetime("ts_event") <= begin_dt);
// var volumeprices = sel_rows.Select(r => Tuple.Create(r.get<int>("volume"), r.get<double>("close"))).ToList();
// double vwap = IndicatorCalc.calculate_VWAP(volumeprices);
// IBlueTableSchema schema = ir.schema.add_columns(Tuple.Create(typeof(double), "vwap"));
// MutableBlueRow nr = new MutableBlueRow(schema, ir);
// //vwap = -313.987;
// nr.setValue("vwap", vwap);
// return nr;
// });
// return rs;
// }
// // the args want dt1 > dt2, if its nto we, swap them
// static void swap_if_necessary(ref DateTime dt1, ref DateTime dt2) {
// if (dt1 < dt2) {
// DateTime aux = dt1;
// dt1 = dt2;
// dt2 = aux;
// }
// }
//// [DataStreamUserRespresentation("VWAP")]
//// public class VWAPStream : ViewStream {
//// // returns the VWAP stream for each contract
//// private static ViewStream build_vwap_stream(string symbol, DateTime begin_dt, DateTime end_dt) {
//// var ohlcv_params = OHLCVStream.get_params(begin_dt, end_dt);
//// ViewStream ohlc_str = new OHLCVStream(symbol, ohlcv_params);
//// // stream is grouped per contract and then vwap is calculated per contract
//// GlobalMapStream rs = new GlobalMapStream(ohlc_str, "ts_event", "contract",
//// (rows, idx) => {
//// Log.Gen.info($"VWAPStream.GlobalMapStream delegate n_rows: {rows.Count} idx: {idx}");
//// IBlueRow ir = rows.ElementAt((int)idx);
//// var sel_rows = rows.Skip((int)idx).Where(r => r.get_datetime("ts_event") >= end_dt && r.get_datetime("ts_event") <= begin_dt);
//// var volumeprices = sel_rows.Select(r => Tuple.Create(r.get<int>("volume"), r.get<double>("close"))).ToList();
//// double vwap = IndicatorCalc.calculate_VWAP(volumeprices);
//// IBlueTableSchema schema = ir.schema.add_columns(Tuple.Create(typeof(double), "vwap"));
//// MutableBlueRow nr = new MutableBlueRow(schema, ir);
//// //vwap = -313.987;
//// nr.setValue("vwap", vwap);
//// return nr;
//// });
//// return rs;
//// }
//// // the args want dt1 > dt2, if its nto we, swap them
//// static void swap_if_necessary(ref DateTime dt1, ref DateTime dt2) {
//// if (dt1 < dt2) {
//// DateTime aux = dt1;
//// dt1 = dt2;
//// dt2 = aux;
//// }
//// }
// // this assumes the ohlc_str has diff contracts
// public VWAPStream(string symbol_or_contract, DateTime begin_dt, DateTime end_dt) {
// swap_if_necessary(ref begin_dt, ref end_dt);
// Tuple<ContractUtils.ContractCategory, string> ccat_sym_contr = ContractUtils.get_symbol_category(symbol_or_contract);
// string symbol = String.Empty;
// symbol = ccat_sym_contr.Item2;
// var ohlcv_params = OHLCVStream.get_params(begin_dt, end_dt);
// ViewStream vwap_str = build_vwap_stream(symbol, begin_dt, end_dt);
// // vwap_str now has vwap for all contracts. if the arg was a symbol, we need to filter it for the active contract,
// // whihc we do with FutureActivesContractStream. If the arg was a contract we'll simply filter it
// ViewStream rs = vwap_str;
// switch (ccat_sym_contr.Item1) {
// case ContractUtils.ContractCategory.FutureString:
// case ContractUtils.ContractCategory.SpreadString:
// rs = new FilterStream(vwap_str, r => r.get<string>("contract").CompareTo(ccat_sym_contr.Item2) == 0);
// break;
// case ContractUtils.ContractCategory.Symbol:
// rs = new FuturesActiveContractStream(vwap_str);
// break;
// }
// rs = new FilterStream(rs, r => r.get_datetime("ts_event").CompareTo(begin_dt) == 0);
// rs = new NewProjectStream(rs, "! vwap");
// mySetRealSource(rs);
// }
//// // this assumes the ohlc_str has diff contracts
//// public VWAPStream(string symbol_or_contract, DateTime begin_dt, DateTime end_dt) {
//// swap_if_necessary(ref begin_dt, ref end_dt);
//// Tuple<ContractUtils.ContractCategory, string> ccat_sym_contr = ContractUtils.get_symbol_category(symbol_or_contract);
//// string symbol = String.Empty;
//// symbol = ccat_sym_contr.Item2;
//// var ohlcv_params = OHLCVStream.get_params(begin_dt, end_dt);
//// ViewStream vwap_str = build_vwap_stream(symbol, begin_dt, end_dt);
//// // vwap_str now has vwap for all contracts. if the arg was a symbol, we need to filter it for the active contract,
//// // whihc we do with FutureActivesContractStream. If the arg was a contract we'll simply filter it
//// ViewStream rs = vwap_str;
//// switch (ccat_sym_contr.Item1) {
//// case ContractUtils.ContractCategory.FutureString:
//// case ContractUtils.ContractCategory.SpreadString:
//// rs = new FilterStream(vwap_str, r => r.get<string>("contract").CompareTo(ccat_sym_contr.Item2) == 0);
//// break;
//// case ContractUtils.ContractCategory.Symbol:
//// rs = new FuturesActiveContractStream(vwap_str);
//// break;
//// }
//// rs = new FilterStream(rs, r => r.get_datetime("ts_event").CompareTo(begin_dt) == 0);
//// rs = new NewProjectStream(rs, "! vwap");
//// mySetRealSource(rs);
//// }
// public VWAPStream(string symbol_or_contract, string begin_dt, string end_dt)
// :
// this(symbol_or_contract, DateTime.Parse(begin_dt), DateTime.Parse(end_dt)) { }
// }
// }
//// public VWAPStream(string symbol_or_contract, string begin_dt, string end_dt)
//// :
//// this(symbol_or_contract, DateTime.Parse(begin_dt), DateTime.Parse(end_dt)) { }
//// }
//// }
using System;
//using System;
using BCG;
//using BCG;
using bts.api;
//using bts.api;
namespace bts.SilverPine {
/*
----------------------------------------------------------------------
*Expression*
----------------------------------------------------------------------
//nl("Find Settlement price of the SP500 future in the first week of 2023.")
//namespace bts.SilverPine {
// /*
// ----------------------------------------------------------------------
// *Expression*
// ----------------------------------------------------------------------
// //nl("Find Settlement price of the SP500 future in the first week of 2023.")
r=FutureHistoricalBars("ES", "One_Day", "1/1/2024", "2/1/2024")
r=math(r, "vprice = volume * close")
r=aggregate(r, "vprice:Sum, volume:Sum")
r=math(r, "vwap = vprice_sum / volume_sum")
r=newProject(r, "vwap !")
// r=FutureHistoricalBars("ES", "One_Day", "1/1/2024", "2/1/2024")
// r=math(r, "vprice = volume * close")
// r=aggregate(r, "vprice:Sum, volume:Sum")
// r=math(r, "vwap = vprice_sum / volume_sum")
// r=newProject(r, "vwap !")
----------------------------------------------------------------------
*Schema*
----------------------------------------------------------------------
id System.Int32
vwap System.Double
*/
// ----------------------------------------------------------------------
// *Schema*
// ----------------------------------------------------------------------
// id System.Int32
// vwap System.Double
// */
[DataStreamUserRespresentation("VWAP")]
public class VWAPStream : ViewStream {
public VWAPStream(string symbol_or_contract, DateTime beg_dt, DateTime end_dt)
:
this(symbol_or_contract, beg_dt, end_dt, ViewStreamDatabase.instance) { }
// [DataStreamUserRespresentation("VWAP")]
// public class VWAPStream : ViewStream {
// public VWAPStream(string symbol_or_contract, DateTime beg_dt, DateTime end_dt)
// :
// this(symbol_or_contract, beg_dt, end_dt, ViewStreamDatabase.instance) { }
public VWAPStream(string symbol_or_contract, DateTime beg_dt, DateTime end_dt, ViewStreamDatabase db) {
//var fc = new ContractUtils.FutureContract(symbol_or_contract);
ViewStream r = new FutureHistoricalBarsStream(symbol_or_contract, beg_dt.ToString(), end_dt.ToString());
// since this is already filtered, we can groupBy iit without a groupBy column
r = new MathStream(r,
"vprice = volume * close");
r = new NonIncGroupByStream(r,
"vprice:Sum, volume:Sum");
r = new MathStream(r,
"vwap = vprice_sum / volume_sum");
r = new NewProjectStream(r,
"vwap !");
mySetRealSource(r);
}
}
}
// public VWAPStream(string symbol_or_contract, DateTime beg_dt, DateTime end_dt, ViewStreamDatabase db) {
// //var fc = new ContractUtils.FutureContract(symbol_or_contract);
// ViewStream r = new FutureHistoricalBarsStream(symbol_or_contract, beg_dt.ToString(), end_dt.ToString());
// // since this is already filtered, we can groupBy iit without a groupBy column
// r = new MathStream(r,
// "vprice = volume * close");
// r = new NonIncGroupByStream(r,
// "vprice:Sum, volume:Sum");
// r = new MathStream(r,
// "vwap = vprice_sum / volume_sum");
// r = new NewProjectStream(r,
// "vwap !");
// mySetRealSource(r);
// }
// }
//}
@@ -1,53 +1,53 @@
using BCG;
//using BCG;
namespace bts.SilverPine {
/*
----------------------------------------------------------------------
*Expression*
----------------------------------------------------------------------
//r=FutureHistoricalBars("ES*", "02/01/2024 09:30 AM", "02/01/2024 09:30 PM")
r=FutureHistoricalBars("ES*","One_Day", 02/01/2024, 02/01/2025)
r=contractExpander(r,"symbol")
r=filter(r,"contract_type='FutureSpread'")
r=aggregate(r,"symbol","volume:Sum","ts_event")
r=aggregate(r,"volume_sum:Max")
r=newProject(r,"volume_sum_max:volume symbol !")
//namespace bts.SilverPine {
// /*
// ----------------------------------------------------------------------
// *Expression*
// ----------------------------------------------------------------------
// //r=FutureHistoricalBars("ES*", "02/01/2024 09:30 AM", "02/01/2024 09:30 PM")
// r=FutureHistoricalBars("ES*","One_Day", 02/01/2024, 02/01/2025)
// r=contractExpander(r,"symbol")
// r=filter(r,"contract_type='FutureSpread'")
// r=aggregate(r,"symbol","volume:Sum","ts_event")
// r=aggregate(r,"volume_sum:Max")
// r=newProject(r,"volume_sum_max:volume symbol !")
----------------------------------------------------------------------
*Schema*
----------------------------------------------------------------------
symbol System.String
id System.Int32
volume System.Double
*/
// ----------------------------------------------------------------------
// *Schema*
// ----------------------------------------------------------------------
// symbol System.String
// id System.Int32
// volume System.Double
// */
[DataStreamUserRespresentation("VolumeAgg")]
public class VolumeAggregationStream : ViewStream {
public VolumeAggregationStream(string future_contract, string contract_type, DateTime range_beg, DateTime range_end)
: this(future_contract, contract_type, range_beg, range_end, ViewStreamDatabase.instance) { }
// [DataStreamUserRespresentation("VolumeAgg")]
// public class VolumeAggregationStream : ViewStream {
// public VolumeAggregationStream(string future_contract, string contract_type, DateTime range_beg, DateTime range_end)
// : this(future_contract, contract_type, range_beg, range_end, ViewStreamDatabase.instance) { }
public VolumeAggregationStream(string future_contract, string contract_type, string range_beg, string range_end)
:
this(future_contract, contract_type, DateTime.Parse(range_beg), DateTime.Parse(range_end), ViewStreamDatabase.instance) { }
// public VolumeAggregationStream(string future_contract, string contract_type, string range_beg, string range_end)
// :
// this(future_contract, contract_type, DateTime.Parse(range_beg), DateTime.Parse(range_end), ViewStreamDatabase.instance) { }
public VolumeAggregationStream(string future_contract, string contract_type, DateTime range_beg, DateTime range_end, ViewStreamDatabase db) {
future_contract = future_contract.EndsWith("*") ? future_contract : future_contract + "*";
ViewStream r;
r = new FutureHistoricalBarsStream(future_contract,
range_beg.ToString(), range_end.ToString());
r = new ContractExpanderStream(r, "symbol");
r = new FilterStream(r, $"contract_type='{contract_type}'");
r = new NonIncGroupByStream(r, "symbol", "volume:Sum", "ts_event");
r = new NonIncGroupByStream(r, "volume_sum:Max");
r = new NewProjectStream(r, "volume_sum_max:volume symbol !");
mySetRealSource(r);
}
}
// public VolumeAggregationStream(string future_contract, string contract_type, DateTime range_beg, DateTime range_end, ViewStreamDatabase db) {
// future_contract = future_contract.EndsWith("*") ? future_contract : future_contract + "*";
// ViewStream r;
// r = new FutureHistoricalBarsStream(future_contract,
// range_beg.ToString(), range_end.ToString());
// r = new ContractExpanderStream(r, "symbol");
// r = new FilterStream(r, $"contract_type='{contract_type}'");
// r = new NonIncGroupByStream(r, "symbol", "volume:Sum", "ts_event");
// r = new NonIncGroupByStream(r, "volume_sum:Max");
// r = new NewProjectStream(r, "volume_sum_max:volume symbol !");
// mySetRealSource(r);
// }
// }
[DataStreamUserRespresentation("PeriodicVolumeAgg")]
public class PeriodicVolumeAggregationStream : ViewStream {
public PeriodicVolumeAggregationStream(string future_contract, string contract_type, DateTime range_beg, DateTime range_end,
TimeSpan period, ViewStreamDatabase db) {
}
}
}
// [DataStreamUserRespresentation("PeriodicVolumeAgg")]
// public class PeriodicVolumeAggregationStream : ViewStream {
// public PeriodicVolumeAggregationStream(string future_contract, string contract_type, DateTime range_beg, DateTime range_end,
// TimeSpan period, ViewStreamDatabase db) {
// }
// }
//}
@@ -76,30 +76,3 @@ namespace bts.SilverPine {
}
}
}
namespace bts.SilverPine {
using BCG;
using bts.api;
using Newtonsoft.Json.Serialization;
[DataStreamUserRespresentation("FuturesContinousContract")]
public class FuturesContinousContractStream : ViewStream {
// the argument is symbol, like "ES" etc and not the contract itself
public FuturesContinousContractStream(string symbol) : this(symbol, ViewStreamDatabase.instance) { }
public FuturesContinousContractStream(string symbol, ViewStreamDatabase db) {
symbol = symbol.EndsWith("*") ? symbol : symbol + "*";
ViewStream r = new FuturesHistorical1DayBarsStream(symbol);
r = new ContractExpanderStream(r, "symbol");
r = new FilterStream(r, "contract_type='Future'");
// finding the max volume across symbols, if we need to do this for open_interest, then join here
// with the open interest stream on symbol/contract
r = new NonIncGroupByStream(r, "ts_event", "volume:Max");
r = new NewProjectStream(r, "ts_event symbol !");
r = new SortStream(r, "-ts_event");
mySetRealSource(r);
}
}
}
@@ -1,16 +1,18 @@
namespace bts.SilverPine {
using bts;
using BCG;
// it uses the volume (better to use open interest perhap) to detect the rollover, ie when a new
// future exp becomes the highest volume one
[DataStreamUserRespresentation("FuturesRoll")]
public class FuturesRollStream : ViewStream {
public FuturesRollStream(string future_symbol) {
ViewStream rs = new FuturesContinousContractStream(future_symbol);
rs = new ColumnEvolutionStream(rs, "ts_event", "symbol");
rs = new NewProjectStream(rs, "symbol ts_event:roll_date !");
mySetRealSource(rs);
}
}
}
//TODO: Remove
//namespace bts.SilverPine {
// using bts;
// using BCG;
// // it uses the volume (better to use open interest perhap) to detect the rollover, ie when a new
// // future exp becomes the highest volume one
// [DataStreamUserRespresentation("FuturesRoll")]
// public class FuturesRollStream : ViewStream {
// public FuturesRollStream(string future_symbol) {
// ViewStream rs = new FuturesContinousContractStream(future_symbol);
// rs = new ColumnEvolutionStream(rs, "ts_event", "symbol");
// rs = new NewProjectStream(rs, "symbol ts_event:roll_date !");
// mySetRealSource(rs);
// }
// }
//}
@@ -1,26 +1,27 @@
namespace bts.SilverPine {
using BCG;
// TODO: Remove
//namespace bts.SilverPine {
// using BCG;
using bts;
// used directly in blueorb
[DataStreamUserRespresentation("Settlement")]
public class SettlementPriceStream : ViewStream {
// using bts;
// // used directly in blueorb
// [DataStreamUserRespresentation("Settlement")]
// public class SettlementPriceStream : ViewStream {
// public SettlementPriceStream(string future_name, string date_txt)
// : this(future_name, date_txt, date_txt) { }
// // public SettlementPriceStream(string future_name, string date_txt)
// // : this(future_name, date_txt, date_txt) { }
// public SettlementPriceStream(string future_name, string begin_dt, string end_dt)
// : this(future_name, DateTime.Parse(begin_dt).Date, DateTime.Parse(end_dt).Date) { }
// // public SettlementPriceStream(string future_name, string begin_dt, string end_dt)
// // : this(future_name, DateTime.Parse(begin_dt).Date, DateTime.Parse(end_dt).Date) { }
public SettlementPriceStream(string future_contract, DateTime begin_dt)
: this(future_contract, begin_dt, begin_dt) { }
// public SettlementPriceStream(string future_contract, DateTime begin_dt)
// : this(future_contract, begin_dt, begin_dt) { }
public SettlementPriceStream(string future_contract, DateTime begin_dt, DateTime end_dt) {
ContractUtils.FutureContract fc = new ContractUtils.FutureContract(future_contract);
ViewStream rs = new FutureHistoricalBarsStream(fc, BarSize.ONE_DAY, new DateTimeRange(begin_dt, end_dt));
rs = new NewProjectStream(rs, "symbol ts_event:date close:settlement !");
mySetRealSource(rs);
}
// public SettlementPriceStream(string future_contract, DateTime begin_dt, DateTime end_dt) {
// ContractUtils.FutureContract fc = new ContractUtils.FutureContract(future_contract);
// ViewStream rs = new FutureHistoricalBarsStream(fc, BarSize.ONE_DAY, new DateTimeRange(begin_dt, end_dt));
// rs = new NewProjectStream(rs, "symbol ts_event:date close:settlement !");
// mySetRealSource(rs);
// }
}
}
// }
//}
@@ -1,276 +1,571 @@
# **DSL Master Learning Prompt**
# BlueOrbDSL Master Learning Prompt v1.1 (Optimized)
This prompt is designed to teach the foundational concepts of the DSL for financial data streaming and transformation. The DSL operates on "streams," which fetch, transform, and process financial market data in real-time or historical contexts. Forget everything I have taught you about this DSL until now (since there are some contradicting and breaking changes that can not be reconciled with anything you might have learned before) and re-learn from this prompt.
**This version is the *final, frozen* canonical Master Learning Prompt for BlueOrbDSL model training and generation.**
**No further edits are required. This is the single source of truth for Learning Mode.**
**Reviewed and validated as optimal on \[insert date here, e.g. June 9, 2025\] after full Meta Mode review.**
# **Introduction to Streams**
**Change Log — v1 → v1.1**
(June 7, 2025\)
The DSL operates on "streams," its core functional units. Streams fetch tabular (from database, text files or network) or transform/operate on tabular data. They either act as **Datasources** (e.g., FutureHistoricalBars), that bootstrap the data or **Operators** (e.g., `FilterStream, ProjectStream, SingleRowMathStream, GroupByStream` etc) that transform or filter tabular data. Though the DSL can work on any tabular data, it is primarily written for streaming or static financial data
✅ Applied full Meta Mode review recommendations:
# **Streams**
- Standardized **Arguments** section (bullet lists, first arg \= `parent_stream` for operators)
- Added **Adds Columns?** and **Modifies Schema?** rows for *all* streams
- Clarified **Special Notes** for all streams
- Clearly stated UnionStream schema requirements
- Explicitly documented JoinStream as inner join, with left/right prefix behavior
- Clarified behavior of **TimeBasedRollingFilterStream**
- Changed DateTimeExpanderStream new columns to a bullet list for clarity
- Standardized all **Example Usage** to fenced `dsl` code blocks
- Shortened **PythonDataframeStream** example; big one should go to appendix if needed
- Minor language and consistency cleanups across stream definitions
## **FutureHistoricalBars (DataSource Stream)**
The `FutureHistoricalBars` stream fetches OHLCV (Open, High, Low, Close, Volume) bars for financial futures contracts. It acts as a **data source** and provides the initial table that can be further processed by operators.
#### **Stream arguments (parameters):**
1. **`Future_contract` (string):** Specifies the contract to fetch data for. This can be one of the following 3 types
1. **Continuous contracts:** Use the base symbol (e.g., `ES` for S\&P 500 futures, CL for WTI Crude Oil Futures, GC for Gold Futures etc).
2. **Specific contracts:** Use a detailed symbol (e.g., `ESZ9` for December 2029 S\&P 500 futures, `GCU9`, for September 2025 Gold Future, `SIH9-J9`,
For SIlver March April 2019 Silver Future Calendar spread)
3. **All Contracts for a symbol:** If instead of specifying individual contracts, data for all contracts for a specific symbol is needed, use the symbol name followed by *.* So `ES*` for all SP Futures and spread contracts, ZC\* for all corn contracts etc.
2. **`Bar_size` (string):** The second argument defines the granularity of the bars. Accepted values:
* `ONE_DAY` – Daily bars.
* `ONE_MIN` – Minute bars.
3. **Optional:**
* **`range_begin` (datetime):** Start of the desired time range.
* **`range_end` (datetime):** End of the desired time range.
#### **Schema:**
The output of `FutureHistoricalBars` produces a table with the following schema (this is a classic OHLCV schema that you can find online easily)
| id | ts\_event | contract | open | high | low | close | volume |
| :--- | :-------------------- | :------- | :------ | :------ | :------ | :------ | :----- |
| `0` | `2024-06-08T00:00:00` | `ESZ9` | 5001.50 | 5052.25 | 4990.75 | 5025.50 | 866000 |
| | | | | | | | |
# **FilterStream (Operator Stream)**
Filters rows of a parent stream based on conditions applied to its columns.
#### **Stream arguments (parameters):**
1. `parent_stream (another stream)` This stream provides the data, that the FilterStream subscribers to
2. `filter_condition (string)` This is a text expressing a boolean condition that is applied to each row of the parent stream, if it returns false, the row is filtered out
#### **Examples:**
1. **NL prompt:** Show me all the days when the open price for SP 500 Futures was greater than or equal to 5000:
**DSL:** `r = FilterStream(FutureHistoricalBars("ES", "ONE_DAY"), "open >= 5000")`
2. **NL prompt:** Show me all the days when the close price for SP 500 Futures was less than 3000 and volume was greater than half a million
**DSL:** `r = FilterStream(FutureHistoricalBars("ES", "ONE_MIN"), "close < 3000 and volume > 500000)`
3. **NL prompt:** Show me all the daily bars for the SP 500 Future spread ESU9-Z9 where its volume exceeded 100K
**DSL:** `r = FilterStream(FutureHistoricalBars("ES*", "ONE_MIN"), "contract = ‘ESU9-Z9’ and volume >= 100000”)`
**Notice** how expressions can be joined with AND and OR. These expressions can be nested with parenthesis.Also string needs to be in single quotes.
# **ProjectStream (Operator Stream)**
Changes the name of the columns in the parent stream. So for example, if we have to rename the close column to settlement, we can do this with ProjectStream
#### **Stream arguments (parameters):**
3. `parent_stream (another stream)` This stream provides the data, that the ProjectStream subscribers to
4. `renaming (string)` This is a text expressed in the form original\_col\_name:projected\_col\_name, original\_col\_name:projected\_col\_name,
#### **Examples:**
The following DSL renames the open column to O, high to H, low to L and close to C
**DSL:** `r = ProjectStream(FutureHistoricalBars("ES", "ONE_DAY"), "open:O, high:H, low:L, close:C”)`
# **GroupByStream (Operator Stream)**
The `GroupByStream` function is a DSL operator used to perform aggregation operations on rows of a parent stream. It works similarly to SQL’s `GROUP BY`, allowing rows to be grouped by a specific column (optional) and applying aggregation functions to other columns.
**This version is now ready to serve as the authoritative MLP for BlueOrbDSL training and generation.**
When instructed to "Switch to Learning Mode," this will become the model’s single source of truth.
---
#### **Stream arguments (parameters):**
# Table of Contents
1. **parent\_stream (another stream):** The source stream provides the data to be operated on
2. **Group By Column (string, optional):**
A column in the schema of the parent stream used to group rows. All rows with the same value in this column are aggregated into a single group. If omitted, the aggregation is performed over all rows of the parent stream.
3. **Operation Text (string):** A string defining the column(s) to be aggregated and the operation(s) to apply, in the format:
`<column_name: operation>, <column_name: operation>, …` Each `<column_name: operation>` pair produces a new column in the output stream. The new column name is formatted as `<column_name>_<operation>` in lowercase.
1. Preamble
2. Streams Overview
3. Data Sources
3.1 FutureHistoricalBarsStream
3.2 LiveBBOStream
3.3 LiteralStream
4. Operators
4.1 FilterStream
4.2 ProjectStream
4.3 GroupByStream
4.4 SingleRowMathStream
4.5 MultiRowRollingMathStream
4.6 UnionStream
4.7 JoinStream
4.8 DateTimeExpanderStream
4.9 ContractExpanderStream
4.10 CountStream
4.11 TimeSeriesStream
4.12 TimeBasedRollingFilterStream
4.13 PythonDataframeStream
5. Reserved Words & Enumerations
6. Common Pitfalls
---
#### **Valid Aggregation Operations:**
## 1 Preamble
* `Mean`: Arithmetic mean of the values.
* `Median`: Median of the values.
* `Mode`: Most frequently occurring value.
* `Range`: Difference between the max and min.
* `StdDev`: Standard deviation of the values.
* `Max`: Maximum value.
* `Min`: Minimum value.
* `First`: First value in the group.
* `Last`: Last value in the group.
* `Sum`: Sum of the values.
* `EdgeDiff`: Difference between the edge values, meaning the difference between a certain col belonging to the first and last rows respectively
* `EdgeDiffPercentage`: Same as `EdgeDiff` but expressed as percentage instead
BlueOrbDSL turns natural-language finance questions into **stream pipelines** that operate on tabular market data (historical or real-time).
Each *stream* is either a **Data Source** (fetches data) or an **Operator** (transforms data).
Streams compose: the output table of one becomes the input of the next.
**Convention:** For all operators, the **first argument** is always the `parent_stream`. Subsequent arguments are constructor parameters.
---
#### **Examples:**
## 2 Streams Overview
1. **NL:**
Find the total SP 500 volume executed in the first week of 2021\.
**DSL:**
`bars = FutureHistoricalBars("ES", “ONE_DAY”, “2021-01-03T00:00:00”,“2021-01-11T00:00:00”)`
`result = GroupByStream(bars,”volume_Sum”)`
2. **NL:**
**Find the highest price and lowest price of the WTI Crude Oil futures over the first half of 2024**
**DSL:**
`bars = FutureHistoricalBars("CL", “ONE_DAY”, “2021-01-03T00:00:00”,“2021-01-06T00:00:00”)`
`result = GroupByStream(bars,”high:Max low:Min”)`
`result = ProjectStream(result,”high_max:highest_price min_low: lowest_price”)`
3. **NL:**
**Find the Average daily volume per contract in Natural Gas in 2023**
**DSL:**
`bars = FutureHistoricalBars("NG*", “ONE_DAY”, “2023-01-03T00:00:00”,“2023-31-12T23:59:59”)`
`result = GroupByStream(bars, “contract”, ”volume:Mean”)`
`result = ProjectStream(result,”mean_volume:average_daily_volume”)`
All stream definitions follow this table layout:
| Field | Meaning |
| :------------------- | :----------------------------------------------------- |
| **Type** | DataSource / Operator |
| **Purpose** | One-sentence description |
| **Arguments** | Bullet list; first arg \= parent\_stream (if Operator) |
| **Output Schema** | Columns created or removed |
| **Adds Columns?** | Yes / No |
| **Modifies Schema?** | Yes / No |
| **Example Usage** | Fenced `dsl` block |
| **Special Notes** | Constraints / caveats |
---
#### **Schema Impact:**
## 3 Data Sources
The `AggregateStream` modifies the schema of the parent stream by adding new columns for each aggregation operation. Existing columns are unaffected.
### 3.1 FutureHistoricalBarsStream
# **SingleRowMathStream (Operator Stream)**
| Field | Description |
| :------------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Type** | DataSource |
| **Purpose** | Fetch historical OHLCV bars for futures contracts |
| **Arguments** | • `contract` — continuous (`ES`), specific (`ESZ9`), or wildcard (`ES*`) • \``bar_size`\` — ONE\_DAY, ONE\_MIN • `` `range_begin` `` — ISO datetime, optional • `` `range_end` `` — ISO datetime, optional |
| **Output Schema** | `id, ts_event, date, contract, open, high, low, close, volume` |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | No |
| **Example Usage** | `bars = FutureHistoricalBarsStream("ES", "ONE_DAY", "2023-01-01", "2023-12-31")` |
| **Special Notes** | Whereas the ts\_event column is the timestamp of the column, date is the date of the day. |
The `SingleRowMathStream` is a DSL operator used to perform math operations, one row at a time, on the parent stream.
### 3.2 LiveBBOStream
| Field | Description |
| :------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Type** | DataSource |
| **Purpose** | Stream live best-bid/offer quotes |
| **Arguments** | • `contract` — e.g. `"NGU5"` or `"CL*"` |
| **Output Schema** | `ts_event, contract, bid_price, bid_size, ask_price, ask_size` |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | No |
| **Example Usage** | `bbo = LiveBBOStream("NGU5")` |
| **Special Notes** | LiveBBOStream produces the latest value of the bid/ask price and size. It does not maintain history. Each new row overwrites the old one. For history use TimeSeriesStream |
### 3.3 LiteralStream
| Field | Description |
| :------------------- | :---------------------------------------------------------------------------------------------------------------- |
| **Type** | DataSource |
| **Purpose** | Inject small static tables via JSON records |
| **Arguments** | • `json_text` — string |
| **Output Schema** | Inferred from JSON keys |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | No |
| **Example Usage** | `meta = LiteralStream('[{"event":"OPEC","date":"2024-06-01"}]')` |
| **Special Notes** | If you download the data from the internet, and want to use it conjunction with the BlueOrbDSL this can be useful |
---
#### **Stream arguments (parameters):**
## 4 Operators
1. **parent\_stream (another stream):**The source stream provides the base data
2. **math\_expression(string):** This is a math expression which allows the creation of one or more new columns that are the output of a math expression on the existing columns provided by parent\_stream. This can a simple math operation like new\_col \= a \+ b, or any function from C\#’s System.Math namespace, like new\_column \= Math.Log(Math.Pow(volume, 1.1))
### 4.1 FilterStream
#### **Examples:**
| Field | Description |
| :------------------- | :--------------------------------------------------------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Keep rows where a boolean condition is true |
| **Arguments** | • `parent_stream` •`` `condition` `` — string |
| **Output Schema** | Same as input |
| **Adds Columns?** | No |
| **Modifies Schema?** | No |
| **Special Notes** | Condition syntax: use ASCII single quotes (`contract = 'ESZ9'`). Supports `and`, `or`, parentheses, equality and inequality. |
4. **NL:**
Find the daily price movement in the SP 500 Futures through the month of Jan
**DSL:**
`bars = FutureHistoricalBars("ES", “ONE_DAY”, “2024-01-01T00:00:00”,“2024-01-31T00:00:00”)`
`result = SingleRowMathStream(bars,”daily_move=close-open”)`
5. **NL:**
Find the daily typical price of the WTI Crude Oil futures over the first half of 2024
**DSL:**
`bars = FutureHistoricalBars("CL", “ONE_DAY”, “2021-01-03T00:00:00”,“2021-01-06T00:00:00”)`
`result = SingleRowMathStream(bars, “typical_price = (high+low+close)/3.0”)`
### 4.2 ProjectStream
#### **Input Space (Scope):** The scope of this operation is a **single row.** The SingleRowMAthStream as the name signifies, operates on columns of the same row only. **No multi row operations are allowed.** For multi row operations using what follows.
**Schema Impact:**
# **MultiRowRollingMathStream (Operator Stream)**
The `MultiRowRollingMathStream` function is a DSL operator used to perform mostly math operations on a rolling set of rows of a parent stream. It works similarly to the GroupByStream (described above) but instead of grouping rows together based on a column value, it operates on a rolling set of rows (where the rolling size is provided as an argument).
| Field | Description |
| :------------------- | :---------------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Rename or drop columns |
| **Arguments** | • `parent_stream` • `` `column_map ``\` — string |
| **Output Schema** | Renamed subset |
| **Adds Columns?** | No |
| **Modifies Schema?** | Yes |
| **Example Usage** | `clean = ProjectStream(bars, "close:settlement, volume")` |
| **Special Notes** | **Cannot add new columns or constants.** Use LiteralStream \+ JoinStream if needed. |
---
#### **Stream arguments (parameters):**
### 4.3 GroupByStream
1. **Parent Stream (another stream):** The source stream provides the data to be operated on
2. **Operation Text (string):** A string defining the column(s) to be operated and the operation(s) to apply, in the format:
`<column_name: operation>, <column_name: operation>, …` Each `<column_name: operation>` pair produces a new column in the output stream. The new column name is formatted as `<column_name>_<operation>` in lowercase. This is like the GroupByStream’s operations string.
3. **Sort/Order Column Name(string):** This controls the ordering of the rolling rows. This argument will be in the format \-`<column_name>` to sort the rolling rows in descending order by the `<column_name>`, or \+`<column_name> in` ascending order.
4. **Rolling Window Size(int)(optional):** The size of the rolling window, If this argument is left out all, the operations are applied to all rows, is Global Operation instead of Rolling Window Operation (it's like the optional groupBy col in GroupByStream).
| Field | Description |
| :------------------- | :----------------------------------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Aggregates rows by key (or globally) |
| **Arguments** | • `parent_stream` • `` `group_by` `` — string (blank for global) • `` `aggregation_string` `` — string |
| **Output Schema** | Adds `<column>_<aggregation>` columns |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | Yes |
| **Example Usage** | `daily = GroupByStream(bars, "date", "volume:Sum")` |
| **Special Notes** | None |
### 4.4 SingleRowMathStream
| Field | Description |
| :------------------- | :--------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Add new columns from row-wise math expressions |
| **Arguments** | • `parent_stream` • `` `math_expression` `` — string |
| **Output Schema** | Adds new column(s) |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | No |
| **Example Usage** | `r = SingleRowMathStream(bars, "daily_return = close - open")` |
| **Special Notes** | Operates on one row at a time only. Cannot reference values from other rows. |
### 4.5 MultiRowRollingMathStream
| Field | Description |
| :------------------- | :------------------------------------------------------------------------------------------------------------------------------------------ |
| **Type** | Operator |
| **Purpose** | Aggregates over rolling window of rows |
| **Arguments** | • `parent_stream` • `` `aggregation_string` `` — string • `` `sort_column` `` — string with \+/- prefix • `` `rolling_window_size` `` — int |
| **Output Schema** | Adds `<column>_<aggregation>` columns |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | Yes |
| **Example Usage** | `ma20 = MultiRowRollingMathStream(bars, "close:Mean", "-ts_event", 20)` |
| **Special Notes** | sort\_column should be prefixed with \+ (ascending) or \- (descending). Example: "-ts\_event". |
### 4.6 UnionStream
| Field | Description |
| :------------------- | :------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Append rows from two streams |
| **Arguments** | • `parent_stream1` • `parent_stream2` |
| **Output Schema** | Same as inputs |
| **Adds Columns?** | No |
| **Modifies Schema?** | No |
| **Example Usage** | `unioned = UnionStream(bars1, bars2)` |
| **Special Notes** | **Schemas must match exactly (column names AND types).** |
### 4.7 JoinStream
| Field | Description |
| :------------------- | :-------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Join two streams on equality of selected columns |
| **Arguments** | • `` `left_parent_stream` `` • `` `left_join_cols` `` — comma-separated string • `` `right_parent_stream` `` • `` `right_join_cols` `` — comma-separated string |
| **Output Schema** | Combined columns from both streams |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | Yes |
| **Example Usage** | `joined = JoinStream(a, "ts_event", b, "ts_event")` |
| **Special Notes** | Always performs an **inner join** on equality. If columns overlap, prefix applied (`left_`, `right_`). If join columns args left empty, does cartesian product |
### 4.8 DateTimeExpanderStream
| Field | Description |
| :------------------- | :--------------------------------------------------------------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Adds columns with date/time components from a datetime column |
| **Arguments** | • `parent_stream` • `datetime_column` — string |
| **Output Schema** | Adds: • \`\_date\` • \`\_day\_of\_week\` • \`\_day\_of\_month\` • \`\_month\` • \`\_hour\` • \`\_minute\` • \`\_seconds\_of\_day\` |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | Yes |
| **Example Usage** | `expanded = DateTimeExpanderStream(bars, "ts_event")` |
| **Special Notes** | None |
### 4.9 ContractExpanderStream
| Field | Description |
| :------------------- | :---------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Adds metadata columns based on `contract` |
| **Arguments** | • `parent_stream` • contract\_column — string |
| **Output Schema** | Adds: `• contract_type` |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | Yes |
| **Example Usage** | `expanded = ContractExpanderStream(bars, "contract")` |
| **Special Notes** | None |
### 4.10 CountStream
| Field | Description |
| :------------------- | :-------------------------------- |
| **Type** | Operator |
| **Purpose** | Count number of rows |
| **Arguments** | • `parent_stream` |
| **Output Schema** | `id, count` |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | Yes |
| **Example Usage** | `total_count = CountStream(bars)` |
| **Special Notes** | None |
### 4.11 TimeSeriesStream
| Field | Description |
| :------------------- | :------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Append each update from parent stream into time series |
| **Arguments** | • `parent_stream` • \``buffer_size`\` — int |
| **Output Schema** | Parent schema \+ `timestamp` (if not already present) |
| **Adds Columns?** | Yes |
| **Modifies Schema?** | No |
| **Example Usage** | `ts_bbo = TimeSeriesStream(bbo, 10000)` |
| **Special Notes** | Append-only stream, does not deduplicate or collapse repeated values |
### 4.12 TimeBasedRollingFilterStream
| Field | Description |
| :------------------- | :--------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Keep rows newer than rolling window of N seconds |
| **Arguments** | • `parent_stream` • \``no_of_seconds`\` — int |
| **Output Schema** | Same as input |
| **Adds Columns?** | No |
| **Modifies Schema?** | No |
| **Example Usage** | `recent = TimeBasedRollingFilterStream(trades, 60)` |
| **Special Notes** | Drops rows older than `no_of_seconds` compared to current max timestamp seen |
### 4.13 PythonDataframeStream
| Field | Description |
| :------------------- | :---------------------------------------------------------------------------------------------------------------------- |
| **Type** | Operator |
| **Purpose** | Apply arbitrary Python transformation to DataFrame |
| **Arguments** | • `parent_stream` • `` `python_function` `` — string |
| **Output Schema** | Defined by function |
| **Adds Columns?** | Yes/No — depends on function |
| **Modifies Schema?** | Yes/No — depends on function |
| **Example Usage** | `r = PythonDataframeStream(bars, "def transform(df): return df[df.volume > 1e6]")` |
| **Special Notes** | **Use only as last resort.** Function should be small and well-documented. Avoid large multi-step logic where possible. |
---
#### **Valid Aggregation Operations:**
## 5 Reserved Words & Enumerations
* `Mean`: Arithmetic mean of the values.
* `Median`: Median of the values.
* `Mode`: Most frequently occurring value.
* `Range`: Difference between the max and min.
* `StdDev`: Standard deviation of the values.
* `Max`: Maximum value.
* `Min`: Minimum value.
* `First`: First value in the group.
* `Last`: Last value in the group.
* `Sum`: Sum of the values.
* `EdgeDiff`: Difference between the edge values, meaning the difference between a certain col belonging to the first and last rows respectively
* `EdgeDiffPercentage`: Same as `EdgeDiff` but expressed as percentage instead
- **Bar\_size:** `ONE_DAY`, `ONE_MIN`
- **Aggregation Operations:** `Mean`, `Median`, `Mode`, `Sum`, `Max`, `Min`, `StdDev`, `Range`, `First`, `Last`, `EdgeDiff`, `EdgeDiffPercentage`
---
#### **Examples:**
## 6 Common Pitfalls
1. **NL:**
**Find the 20 d MA of SP 500 Futures close prices**
`bars = FutureHistoricalBars("ES", “ONE_DAY”, “2021-01-03T00:00:00”,“2021-01-11T00:00:00”)`
`result = MultirowRollingMathStream(bars, “close:Mean”, “-ts_event”, 20)`
2. **NL:**
**Find the 90d Standard Deviation of WTI Crude Oil during 2023**
| Pitfall | Correct Approach |
| :---------------------------------------------- | :--------------------------------------------------- |
| Adding literal columns in ProjectStream | Use LiteralStream → JoinStream |
| Smart quotes in conditions | Use `'...'` ASCII quotes |
| Large functions in PythonDataframeStream | Keep minimal or move to appendix |
| Ambiguous TimeBasedRollingFilterStream behavior | Now clearly stated: drops rows older than N seconds. |
**DSL:**
`bars = FutureHistoricalBars("CL", “ONE_DAY”, “2023-01-01T00:00:00”,“2023-12-31T00:00:00”)`
`// calculating typical_price`
## 7 Reinforced Learning DSL Examples
`result = SingleRowMathStream(bars, “typical_price = (high+low+close)/3.0”)`
`result = MultirowRollingMathStream(bars,“typical_price:StdDev”, “-ts_event”, 90)`
This section contains **complete worked examples** of BlueOrbDSL pipelines answering complex natural language queries.
`result=ProjectStream(result, “stddev_typical_price:90dStdDev”)`
Each example shows:
3. **NL:**
**Find the Average daily volume of Natural Gas in 2023**
**DSL:**
`bars = FutureHistoricalBars("NG", “ONE_DAY”, “2023-01-03T00:00:00”,“2023-31-12T23:59:59”)`
`result = MultirowRollingMathStream(bars, ”volume:Mean”, “-ts_event”)`
`result = ProjectStream(result,”mean_volume:average_daily_volume”)`
- The **NL Query** (natural language question)
- The **DSL Solution** in full
- Optional **Notes**
---
#### **Schema Impact:**
### Example 1 — On which dates did SP 500 Futures open 2% higher than the previous day's close?
The `MultirowRollingMathStream` modifies the schema of the parent stream by adding new columns for each operation. Existing columns are unaffected.
#### DSL Solution
# **UnionStream (Operator Stream)**
```
bars = FutureHistoricalBarsStream("ES", "ONE_DAY", "1900-01-01T00:00:00", "2100-12-31T23:59:59")
rolling_bars = MultiRowRollingMathStream(bars, "close:First, open:Last", "+ts_event", 2)
percent_change = SingleRowMathStream(rolling_bars, "percent_change = 100 * (open_last - close_first) / close_first")
filtered_bars = FilterStream(percent_change, "percent_change >= 2")
result = ProjectStream(filtered_bars, "ts_event, percent_change")
```
Like the SQL Union appends all the rows from 2 streams together.
---
#### **Stream arguments (parameters):**
### Example 2 — What is the average price movement for e-minis on Wednesday between 9:30 am and 9:40 AM CST?
5. `parent_stream1 (first stream)` This stream provides part of the data that will will be appended to the other part, provided by the following argument
6. `parent_stream2 (second stream)` This stream provides the other part of the data that will will be appended to the first part, provided by the first argument
#### DSL Solution
#### **Examples:**
```
bars = FutureHistoricalBarsStream("ES", "ONE_MIN", "1900-01-01T00:00:00", "2100-12-31T23:59:59")
expanded_bars = DateTimeExpanderStream(bars, "ts_event")
filtered_bars = FilterStream(expanded_bars, "ts_event_day_of_week = 'Wednesday' and ts_event_hour = 9 and ts_event_minute >= 30 and ts_event_minute < 40")
price_movement = SingleRowMathStream(filtered_bars, "movement = close - open")
average_movement = GroupByStream(price_movement, "", "movement:Mean")
result = ProjectStream(average_movement, "movement_mean:average_price_movement")
```
4. **NL prompt:** Show me all the bars across SP 500 Futures and Nasdaq Futures
**DSL:** `r = UnionStream(FutureHistoricalBars("ES", "ONE_DAY"), FutureHistoricalBars("NQ", "ONE_DAY"))`
---
#### **Schema Impact:**
### Example 3 — What did the volume and price of ZN in the 5 minutes following the last 10 FOMC rates?
None
#### DSL Solution
# **DateTimeExpanderStream (Operator Stream)**
```
bars_1 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2023-11-01T14:00:00", "2023-11-01T14:05:00")
bars_2 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2023-09-20T14:00:00", "2023-09-20T14:05:00")
bars_3 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2023-07-26T14:00:00", "2023-07-26T14:05:00")
bars_4 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2023-06-14T14:00:00", "2023-06-14T14:05:00")
bars_5 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2023-05-03T14:00:00", "2023-05-03T14:05:00")
bars_6 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2023-03-22T14:00:00", "2023-03-22T14:05:00")
bars_7 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2023-02-01T14:00:00", "2023-02-01T14:05:00")
bars_8 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2022-12-14T14:00:00", "2022-12-14T14:05:00")
bars_9 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2022-11-02T14:00:00", "2022-11-02T14:05:00")
bars_10 = FutureHistoricalBarsStream("ZN", "ONE_MIN", "2022-09-21T14:00:00", "2022-09-21T14:05:00")
DateTimExpanderStream expands one datetime column into various datetime components, like seconds in day, day of week, date etc, so new columns are added for those. The existing columns remain as is.
all_bars = UnionStream(bars_1, bars_2)
all_bars = UnionStream(all_bars, bars_3)
all_bars = UnionStream(all_bars, bars_4)
all_bars = UnionStream(all_bars, bars_5)
all_bars = UnionStream(all_bars, bars_6)
all_bars = UnionStream(all_bars, bars_7)
all_bars = UnionStream(all_bars, bars_8)
all_bars = UnionStream(all_bars, bars_9)
all_bars = UnionStream(all_bars, bars_10)
#### **Stream arguments (parameters):**
grouped_data = GroupByStream(all_bars, "date", "open:First, close:Last, high:Max, low:Min, volume:Sum")
result = ProjectStream(grouped_data, "ts_event_date, open_first:5_min_open, close_last:5_min_close, high_max:5_min_high, low_min:5_min_low, volume_sum:5_min_volume")
```
1. `parent_stream1 (base stream)` This parent stream that provides the data, including the datetime columns that will be expanded into multiple columns
2. `Datetime_input_column (string)` This is the column that will be expanded into multiple columns.
---
**New Columns:**
`<input_dt_col>_date` (Date of the value, without the time part)
### Example 4 — How much volume of RBOB typically trades during the settlement window?
* `<input_dt_col>_day,` (Day of the week)
* `<input_dt_col>_day_of_month` (day of the month, This is an int between 0 and 31\)
* `<input_dt_col>_month,` (month of the year, this is a numerical value between 1 and 12
* `input_dt_col>_hour,` (Hour of the day, value bw 0 and 23\)
* `<input_dt_col>_min",` (month part of this DateTime. Value is integer between 1 and 12).
* `<input_dt_col>_seconds_of_day",` (Total seconds of the day)
#### DSL Solution
#### **Examples:**
```
rbob_bars = FutureHistoricalBarsStream("RB", "ONE_MIN", "2023-01-01T00:00:00", "2023-12-31T23:59:59")
expanded_bars = DateTimeExpanderStream(rbob_bars, "ts_event")
settlement_window = FilterStream(expanded_bars, "(ts_event_hour = 14 and ts_event_minute >= 28) or (ts_event_hour = 14 and ts_event_minute < 30)")
typical_settlement_volume = GroupByStream(settlement_window, "ts_event_date", "volume:Sum")
result = ProjectStream(typical_settlement_volume, "ts_event_date, volume_sum:daily_settlement_volume")
```
**DSL:** `r = DateTimeExpanderStream(FutureHistoricalBars("ES", "ONE_DAY"), “ts_event”)`
**Input Schema:**
---
| id | ts\_event | contract | open | high | low | close | volume |
| :--- | :-------- | :------- | :--- | :--- | :--- | :---- | :----- |
### Example 5 — How much SP Futures during the settlement window, average per day, during the 2 week windows before future expiration, in 2023?
**Output Schema:**
#### DSL Solution
| id | ts\_event | contract | open | high | low | close | volume | | | | | | | |
| :--- | :-------- | :------- | :--- | :--- | :--- | :---- | :----- | :--- | :--- | :--- | :--- | :--- | :--- | :--- |
```
sp_bars = FutureHistoricalBarsStream("ES", "ONE_MIN", "2023-01-01T00:00:00", "2023-12-31T23:59:59")
expanded_bars = DateTimeExpanderStream(sp_bars, "ts_event")
settlement_window = FilterStream(expanded_bars, "(ts_event_hour = 15 and ts_event_minute >= 58) or (ts_event_hour = 16 and ts_event_minute < 1)")
filtered_bars = FilterStream(settlement_window, "(ts_event_date >= '2023-03-03' and ts_event_date < '2023-03-17') or (ts_event_date >= '2023-06-02' and ts_event_date < '2023-06-16') or (ts_event_date >= '2023-09-01' and ts_event_date < '2023-09-15') or (ts_event_date >= '2023-12-01' and ts_event_date < '2023-12-15')")
daily_volume = GroupByStream(filtered_bars, "ts_event_date", "volume:Sum")
average_daily_volume = GroupByStream(daily_volume, "", "volume_sum:Mean")
result = ProjectStream(average_daily_volume, "volume_sum_mean:average_daily_volume")
```
---
### Example 6 — Find the difference between the number of days in 2023 that SP 500 went up 2% and when it went down 2%
#### DSL Solution
```
bars = FutureHistoricalBarsStream("ES", "ONE_DAY", "2023-01-01T00:00:00", "2023-12-31T23:59:59")
percentage_change = SingleRowMathStream(bars, "percent_change = 100 * (close - open) / open")
days_up_2_percent = FilterStream(percentage_change, "percent_change >= 2")
days_up_count = CountStream(days_up_2_percent)
days_down_2_percent = FilterStream(percentage_change, "percent_change <= -2")
days_down_count = CountStream(days_down_2_percent)
joined_counts = JoinStream(days_up_count, "", days_down_count, "")
difference_result = SingleRowMathStream(joined_counts, "difference = days_up_count - days_down_count")
result = ProjectStream(difference_result, "difference")
```
---
### Example 7 — Monitor which CL contracts (CL) are trading with bid size \>= 1000, and record the last 1000 events\*
#### DSL Solution
```
bbo = LiveBBOStream("CL*")
large_bid = FilterStream(bbo, "bid_size >= 1000")
large_bid_ts = TimeSeriesStream(large_bid, 1000)
result = ProjectStream(large_bid_ts, "timestamp, ts_event, contract, bid_size, bid_price, ask_price")
```
---
### Example 8 — Identify FutureSpreads in CrudeOil and monitor their spread width
#### DSL Solution
```
bbo = LiveBBOStream("CL*")
expanded_bbo = ContractExpanderStream(bbo, "contract")
future_spreads = FilterStream(expanded_bbo, "contract_type = 'FutureSpread'")
spread_stream = SingleRowMathStream(future_spreads, "spread = ask_price - bid_price")
result = ProjectStream(spread_stream, "ts_event, contract, spread")
```
---
### Example 9 — Track when bid prices in NaturalGas Futures cross above 3.00 and keep only the last 10 seconds of such events
#### DSL Solution
```
bbo = LiveBBOStream("NG")
high_bid = FilterStream(bbo, "bid_price > 3.00")
high_bid_ts = TimeSeriesStream(high_bid, 10000)
recent_high_bid = TimeBasedRollingFilterStream(high_bid_ts, 10)
result = ProjectStream(recent_high_bid, "ts_event, contract, bid_price, bid_size, ask_price")
```
---
### Example 11 — Detect sequences in NaturalGas future where bid\_price increased 3 times consecutively within a rolling 30-second window
#### DSL Solution
```
bbo = LiveBBOStream("NG")
bbo_ts = TimeSeriesStream(bbo, 10000)
recent_bbo = TimeBasedRollingFilterStream(bbo, 30) // 30-second window of updates
result = PythonDataframeStream(recent_bbo, "
def transform(df):
import pandas as pd
df = df.sort_values('ts_event')
df['bid_increase'] = df['bid_price'].diff() > 0
df['consec_increase'] = df['bid_increase'].rolling(window=3).sum()
# Find rows where we just reached 3 consecutive increases
df['trigger'] = (df['consec_increase'] == 3) & (df['consec_increase'].shift(1) < 3)
return df[df['trigger']][['ts_event', 'contract', 'bid_price', 'bid_size']]
")
```
---
### Example 12 — When we get a 5+ level bid sweep in SP 500, show me how the bid behaves in the 10 minutes after — mean, stddev, min/max bid\_price
#### DSL Solution
```
bbo = LiveBBOStream("ES")
bbo_ts = TimeSeriesStream(bbo, 10000)
recent_bbo = TimeSeriesStream(bbo, 5000) // keep large enough window for event tracking
result = PythonDataframeStream(recent_bbo, "
def transform(df):
import pandas as pd
df = df.sort_values('ts_event')
rows = []
in_measurement = False
measurement_start_time = None
measurement_end_time = None
bid_prices_in_window = []
min_bid_size_threshold = 100 # defines what we consider 'stable large bid'
min_levels_stable = 5
prior_bid_sizes = []
for i, row in df.iterrows():
bid_price = row['bid_price']
bid_size = row['bid_size']
ts = pd.to_datetime(row['ts_event'])
# Maintain rolling window of prior bid_sizes
prior_bid_sizes.append(bid_size)
if len(prior_bid_sizes) > min_levels_stable:
prior_bid_sizes.pop(0)
if not in_measurement:
# Detect sweep trigger
if all(x > min_bid_size_threshold for x in prior_bid_sizes[:-1]) and bid_size < min_bid_size_threshold:
# Sweep detected → start measurement window
measurement_start_time = ts
measurement_end_time = measurement_start_time + pd.Timedelta(minutes=10)
bid_prices_in_window = []
in_measurement = True
else:
# Inside measurement window
if ts <= measurement_end_time:
bid_prices_in_window.append(bid_price)
else:
# Window complete → emit stats
if bid_prices_in_window:
mean_bid = pd.Series(bid_prices_in_window).mean()
stddev_bid = pd.Series(bid_prices_in_window).std()
min_bid = pd.Series(bid_prices_in_window).min()
max_bid = pd.Series(bid_prices_in_window).max()
rows.append({
'sweep_time': measurement_start_time,
'measurement_end': measurement_end_time,
'mean_bid_price': mean_bid,
'stddev_bid_price': stddev_bid,
'min_bid_price': min_bid,
'max_bid_price': max_bid
})
# Reset state
in_measurement = False
measurement_start_time = None
measurement_end_time = None
bid_prices_in_window = []
prior_bid_sizes = []
return pd.DataFrame(rows)
")
```
*End of Section 7 — Reinforced Learning DSL Examples*
@@ -0,0 +1,27 @@
**Enter Meta Mode**
You are not learning the DSL yet.
You are reviewing the following **BlueOrbDSL Master Learning Prompt** (MLP) as a *training document for yourself*.
Your goal is to optimize it to make it easier for you — or any future LLM — to learn from it and generate DSL code with high accuracy.
Please analyze the MLP and provide:
1. **Clarity review** — is each stream definition clear, unambiguous, easy to parse?
2. **Consistency review** — do all stream definitions follow the same structure and style?
3. **Terminology check** — are key terms (parent stream, arguments, schema impact, example usage, etc.) used consistently?
4. **Structural feedback** — is the overall document structured in a way that supports learning (headings, bold, tables, code blocks)?
5. **Formatting feedback** — are important warnings / "cannot do X" constraints highlighted clearly?
6. **Missing info** — is anything missing from any stream definition that would help you understand how to use it?
7. **Generation risk spots** — any parts where you foresee yourself making common mistakes due to ambiguity?
**Do not attempt to "learn" this DSL yet.** Stay in Meta Mode and provide feedback as an *editor and optimizer*.
When I later say:
→ **Switch to Learning Mode — this is now the final BlueOrbDSL MLP** → you will forget your Meta comments and treat the MLP as the single authoritative source for DSL generation.
+1 -1
View File
@@ -30,7 +30,7 @@
<PrivateAssets>all</PrivateAssets>
</PackageReference>
<PackageReference Include="Google.Protobuf" Version="3.24.3" />
<PackageReference Include="IronPython" Version="3.4.1" />
<PackageReference Include="IronPython" Version="3.4.2" />
<PackageReference Include="MathNet.Numerics" Version="5.0.0" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.2.0" />
<PackageReference Include="NAudio" Version="2.0.1" />
@@ -113,6 +113,11 @@ class ConcurrentUnorderedMap {
return {data.erase(pos.iter()), mx};
}
size_type erase(const key_type &k) {
std::lock_guard<std::mutex> lock{mx};
return data.erase(k);
}
void clear() {
std::lock_guard<std::mutex> lock{mx};
data.clear();
@@ -77,7 +77,7 @@ MarketDataConfig {
okx {
num_depth_levels: 5
url_rest_base: "https://aws.okx.com"
url_ws_base: "wss://wsaws.okx.com:8443"
url_ws_base: "wss://ws.okx.com:8443"
security_criteria {
clauses {
security_types: SECURITY_TYPE_SPOT
@@ -1,4 +1,5 @@
#include "OkxExecutionGateway.h"
#include "lib/api/service_helpers/AdminServiceUtils.h"
#include "lib/api/services/OrderService.pb.h"
#include "lib/config/protos/okx_exe_cfg.pb.h"
#include "lib/domain_model/Security.pb.h"
@@ -12,6 +13,7 @@
#include "lib/orderman/CancelOrderMsg.h"
#include "lib/orderman/CancelReplaceOrderMsg.h"
#include "lib/orderman/CreateOrderMsg.h"
#include "lib/orderman/events/NewAdminDataEvent.h"
#include "lib/orderman/gateways/IExecutionGateway.h"
#include "lib/orderman/orderman_utils/ExchangePosition.h"
#include "lib/orderman/orderman_utils/ExecutionReport.h"
@@ -168,6 +170,7 @@ struct OkxExecutionGateway::PImpl {
config::get<config::OkxExeCfg>().ws_ping_interval_seconds());
websocket::WebsocketClient ws_client;
tbb::concurrent_queue<std::shared_ptr<ExecutionReport>> buffered_exec_reps;
tbb::concurrent_queue<ExchangePosition> exch_pos_queue;
std::optional<std::thread> ws_thread;
OnExeReport cb;
OnGwStatusChange on_gw_status_change;
@@ -178,6 +181,7 @@ struct OkxExecutionGateway::PImpl {
std::shared_ptr<const ID_GEN> id_gen = std::make_shared<const ID_GEN>();
std::atomic<OkxExecutionState> state = OkxExecutionState::OFF;
ExchangePosition exchange_position;
PImpl(md::SecurityOracle o)
: ws_client(websocket::WebsocketClient(
get_ws_client_config(url, connect_timeout, pong_timeout))),
@@ -310,34 +314,30 @@ struct OkxExecutionGateway::PImpl {
std::optional<std::thread> pos_polling_thread;
std::atomic<bool> pos_polling_running{false};
void position_polling();
void start_pos_polling() {
void start_pos_polling(
const uint32_t account_id, const std::string &account_name) {
if (pos_polling_running)
return;
pos_polling_running = true;
pos_polling_thread.emplace([this]() {
ExchangePosition positions;
pos_polling_thread.emplace([this, account_name, account_id]() {
// ExchangePosition positions;
auto venue = api::VENUE_OKX;
positions.exchange = venue;
auto sleep_ms = 2000ms;
// positions.exchange = venue;
auto sleep_ms = 5000ms;
//create positions request
const std::string REQUEST_PATH = "/api/v5/account/positions";
//data for positions
const std::string account_name =
config::get<config::OkxExeCfg>().account();
while (pos_polling_running) {
http::HttpResponse res;
auto headers = OkxUtils::createHeaders(
api_key, passphrase, secret_key, "GET", REQUEST_PATH,
simulated_trading, "");
ExchangePosition positions = {venue, {}};
http::HttpResponse res;
auto headers = OkxUtils::createHeaders(
api_key, passphrase, secret_key, "GET", REQUEST_PATH,
simulated_trading, "");
http::HttpClient::get(
config::get<config::OkxExeCfg>().rest_url() + REQUEST_PATH,
headers, &res);
LOG(OrderMan, Info, "pos poll response: {}", res.body);
positions.total_exchange_positions.clear();
LOG(OrderMan, Info, "pos poll response: {}", res.body);
auto body = OkxUtils::parseJsonSafely(res.body);
if (body && body->contains("data") && (*body)["data"].is_array()) {
auto &j = *body;
@@ -370,10 +370,15 @@ struct OkxExecutionGateway::PImpl {
avgPx = std::stod(item["avgPx"].get<std::string>());
}
positions.total_exchange_positions.push_back(
{instId, sec, account_name, std::nullopt, pos,
Price(avgPx)});
ExchangePosition::UnitPosition{
instId, sec, account_name, account_id, pos,
Price(avgPx)});
}
if (on_exchange_position_state)
exch_pos_queue.push(std::move(positions));
else {
LOG(OrderMan, Info, "on_exchange_position_state not defined");
}
on_exchange_position_state(positions);
}
std::this_thread::sleep_for(std::chrono::milliseconds(sleep_ms));
}
@@ -490,7 +495,28 @@ struct OkxExecutionGateway::PImpl {
start_impl();
});
}
void send_pos() {
if (exch_pos_queue.empty()) {
return;
}
ExchangePosition pos;
while (exch_pos_queue.try_pop(pos)) {
on_exchange_position_state(std::move(pos));
}
}
void send_exes() {
if (buffered_exec_reps.empty()) {
return;
}
std::shared_ptr<ExecutionReport> er;
while (buffered_exec_reps.try_pop(er)) {
trace::CorID cid = {};
LOG(OrderMan, Info,
"OkxExecutionGateway::dispatch_incoming_fix_messages er: {} ",
er->to_string());
cb(std::move(er), cid);
}
}
void start_impl() {
size_t attempt = 0;
while (state != OkxExecutionState::OFF) {
@@ -685,18 +711,36 @@ void OkxExecutionGateway::modify_order(
OkxUtils::handleOkException("Amend order", err);
}
}
void OkxExecutionGateway::dispatch_incoming_fix_messages() {
if (me_->buffered_exec_reps.empty()) {
void OkxExecutionGateway::on_admin_data(
const bts::mercury::Update<NewAdminDataEvent> &update) {
LOG(OrderMan, Info, "OkxExecutionGateway::on_admin_data t_id: {} t_name: {}",
bts::os::get_thread_id(), bts::os::get_thread_name());
if (update.events == nullptr) {
return;
}
std::shared_ptr<ExecutionReport> er;
while (me_->buffered_exec_reps.try_pop(er)) {
trace::CorID cid = {};
LOG(OrderMan, Info,
"OkxExecutionGateway::dispatch_incoming_fix_messages er: {} ",
er->to_string());
me_->cb(std::move(er), cid);
const auto *account = bts::admin::AdminServiceUtils::find_account(
*update.events->back().admin_data,
config::get<config::OkxExeCfg>().account());
if (account == nullptr) {
LOG(OrderMan, Warn,
"OkxExecutionGateway::on_admin_data Gateway account not found. "
"Cannot execute orders. name: '{}'",
config::get<config::OkxExeCfg>().account());
return;
}
if (config::get<config::OkxExeCfg>().account() != account->name() ||
me_->pos_polling_running) {
return;
}
me_->start_pos_polling(account->id(), account->name());
LOG(OrderMan, Info,
"OkxExecutionGateway::on_admin_data Resolved gateway account '{}', pos "
"polling started",
account->name());
};
void OkxExecutionGateway::dispatch_incoming_fix_messages() {
me_->send_exes();
me_->send_pos();
}
OnExeReport &OkxExecutionGateway::on_exe_report() { return me_->cb; }
OnCancelReject &OkxExecutionGateway::on_cancel_reject() {
@@ -726,11 +770,17 @@ void OkxExecutionGateway::launch_new_order(const CreateOrderMsg &msg) {
auto pos = ord_full_name.find(suffix);
std::string order_type;
switch (msg.time_in_force) {
case api::orderman::TIME_IN_FORCE_IOC: order_type = "ioc"; break;
case api::orderman::TIME_IN_FORCE_FOK: order_type = "fok"; break;
default: order_type = FIX::string_toLower(ord_full_name.substr(pos + suffix.size()));
case api::orderman::TIME_IN_FORCE_IOC:
order_type = "ioc";
break;
case api::orderman::TIME_IN_FORCE_FOK:
order_type = "fok";
break;
default:
order_type =
FIX::string_toLower(ord_full_name.substr(pos + suffix.size()));
}
const std::string ord_type =order_type;
const std::string ord_type = order_type;
const std::string body = fmt::format(
body_template, msg.security->okx().security_desc(), me_->tdmode, side,
@@ -15,9 +15,7 @@ class OkxExecutionGateway : public IExecutionGateway {
void on_securities(const mercury::Update<NewSecuritiesEvent> &) override {
LOG(md, Info,"Not implemented");
};
void on_admin_data(const mercury::Update<NewAdminDataEvent> &) override{
LOG(md, Info,"Not implemented");
};
void on_admin_data(const mercury::Update<NewAdminDataEvent> &) override;
void send_fix(const jetfix::OutFIXPtr &, trace::CorID) override{
LOG(md, Info,"Not implemented");
}
@@ -4,6 +4,8 @@
#include "lib/orderman/orderman_utils/ExecutionReport.h"
#include "lib/utils/string_utils.h"
#include <limits>
namespace binance_execution {
using bts::orderman::ExecutionReport;
@@ -1,10 +1,12 @@
#include "usds_futures_gateway.h"
#include "binance_execution/enums/order_status.h"
#include "binance_execution/enums/time_in_force.h"
#include "binance_execution/execution_report_store.h"
#include "binance_execution/gateway_config.h"
#include "binance_execution/listen_key_requestor.h"
#include "binance_execution/position_calculator.h"
#include "binance_execution/simple_order_key_map.h"
#include "binance_execution/type_converters.h"
#include "binance_execution/type_validators.h"
#include "binance_execution/usds_futures/messages/create_order_request.h"
#include "binance_execution/usds_futures/messages/modify_order_request.h"
@@ -13,6 +15,7 @@
#include "binance_execution/usds_futures/rest_api_client.h"
#include "binance_execution/usds_futures/user_data_stream/user_data_subscription.h"
#include "binance_execution/websocket_api_request.h"
#include "binance_execution/websocket_api_response.h"
#include "lib/api/service_helpers/AdminServiceUtils.h"
#include "lib/api/services/AdminService.pb.h"
#include "lib/api/services/OrderService.h"
@@ -24,6 +27,7 @@
#include "lib/domain_model/Security.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/domain_model/Side.h"
#include "lib/io/JsonReader.h"
#include "lib/io/json_formatter.h"
#include "lib/io/websocket/websocket_client.h"
#include "lib/logging/logging.h"
@@ -103,6 +107,7 @@ using bts::core::StartStopStateMachineBase;
using bts::core::TimeSpan;
using bts::core::TimeStamp;
using bts::io::json_formatter;
using bts::io::JsonReader;
using bts::orderman::CancelOrderMsg;
using bts::orderman::CancelReplaceOrderMsg;
using bts::orderman::CreateOrderMsg;
@@ -178,7 +183,9 @@ struct UsdsFuturesGateway::PImpl {
~PImpl() { stop(); }
enum RequestType {
ORDER,
CREATE_ORDER,
MODIFY_ORDER,
CANCEL_ORDER,
POSITION,
};
@@ -198,6 +205,7 @@ struct UsdsFuturesGateway::PImpl {
SimpleOrderKeyMap key_map;
ConcurrentUnorderedMap<string, RequestType> pending_reqs;
ConcurrentUnorderedMap<string, CreateOrderMsg> pending_create_order_reqs;
ConcurrentQueue<AccountUpdate> incoming_account_updates;
vector<PositionInfo> incoming_position_snapshot;
ConcurrentQueue<OrderUpdate> incoming_order_updates;
@@ -336,8 +344,24 @@ struct UsdsFuturesGateway::PImpl {
});
}
void on_order_ack(const json &msg) {
INFO(R"([binance_ufuts_order_ack.json]{})", msg.dump());
void on_order_ack(
RequestType req_type, const WebsocketApiResponse<nlohmann::json> &res) {
INFO(R"([binance_ufuts_order_ack.json]{})", bts::io::json_formatter(res));
// We generate exe reports from the order update stream (not the ack), with
// a caveat for new-order rejects because they aren't visible in the stream.
// Presumably, they're never actually entered into binance's order sys;
// they're "rejected" before that happens.
if (req_type == CREATE_ORDER && !res.error.is_null()) {
const CreateOrderMsg &bts_req = pending_create_order_reqs[res.id];
shared_ptr<const ExecutionReport> er_p = orders.reject(
bts_req,
fmt::format("Rejected by exchange. {}", res.error.dump()));
const auto &er = *er_p;
orders.erase(er.order_key);
key_map.remove(er.order_key);
internal_order_rejects.push(std::move(er_p));
}
pending_create_order_reqs.erase(res.id);
}
void on_position_snapshot(const json &msg) {
@@ -360,42 +384,42 @@ struct UsdsFuturesGateway::PImpl {
DIAG_FN("on_message", "{}", msg_str);
// Convert to nlohmann::json
json msg;
WebsocketApiResponse<nlohmann::json> res;
try {
msg = json::parse(msg_str);
msg.get_to(res);
} catch (const exception &ex) {
ERROR_FN(
"on_message", "Failed to parse response json: {}.\n{}",
string{ex.what()}, msg_str);
return;
}
auto id_it = msg.find("id");
if (id_it != msg.end()) {
try {
const string id = id_it->get<string>();
auto pr_it = pending_reqs.find(id);
if (pr_it == pending_reqs.end()) {
ERROR(
R"([binance_ufuts_unsolicited_res.json]{{ "id": "{}", "ts": {} }})",
id, bts::core::TimeStamp::now_ns());
return;
}
const RequestType req_type = pr_it->second;
pending_reqs.erase(pr_it);
switch (req_type) {
case ORDER:
on_order_ack(msg);
break;
case POSITION:
on_position_snapshot(msg);
break;
default:
break;
}
} catch (const exception &ex) {
ERROR_FN(
"on_message", "message handler threw. msg: {}; ex: {}", msg_str,
string{ex.what()});
try {
auto pr_it = pending_reqs.find(res.id);
if (pr_it == pending_reqs.end()) {
ERROR(
R"([binance_ufuts_unsolicited_res.json]{{ "id": "{}", "ts": {} }})",
res.id, bts::core::TimeStamp::now_ns());
return;
}
const RequestType req_type = pr_it->second;
pending_reqs.erase(pr_it);
switch (req_type) {
case CREATE_ORDER:
case MODIFY_ORDER:
case CANCEL_ORDER:
on_order_ack(req_type, res);
break;
case POSITION:
on_position_snapshot(msg);
break;
default:
break;
}
} catch (const exception &ex) {
ERROR_FN(
"on_message", "message handler threw. msg: {}; ex: {}", msg_str,
string{ex.what()});
}
}
@@ -415,7 +439,9 @@ struct UsdsFuturesGateway::PImpl {
return;
}
LOG_CREATE_ORDER(Info, m, "ABOUT_TO_SEND");
const auto req = build_request(std::move(params), RequestType::ORDER);
const auto req =
build_request(std::move(params), RequestType::CREATE_ORDER);
pending_create_order_reqs[req.id] = m;
const auto ec = ws_client->send_text(req.to_string());
if (ec) {
LOG_CREATE_ORDER(Error, m, "WS_ERROR", ec.message());
@@ -437,7 +463,8 @@ struct UsdsFuturesGateway::PImpl {
return;
}
LOG_MODIFY_ORDER(Info, m, "ABOUT_TO_SEND");
const auto req = build_request(std::move(params), RequestType::ORDER);
const auto req =
build_request(std::move(params), RequestType::MODIFY_ORDER);
const auto ec = ws_client->send_text(req.to_string());
if (ec) {
LOG_MODIFY_ORDER(Error, m, "WS_ERROR", ec.message());
@@ -458,7 +485,8 @@ struct UsdsFuturesGateway::PImpl {
return;
}
LOG_CANCEL_ORDER(Info, m, "ABOUT_TO_SEND");
const auto req = build_request(std::move(params), RequestType::ORDER);
const auto req =
build_request(std::move(params), RequestType::CANCEL_ORDER);
const auto ec = ws_client->send_text(req.to_string());
if (ec) {
LOG_CANCEL_ORDER(Error, m, "WS_ERROR", ec.message());
@@ -0,0 +1,44 @@
#include "websocket_api_response.h"
namespace binance_execution {
template <typename Result>
std::string WebsocketApiResponse<Result>::to_string() const {
std::string str = fmt::format("{}", *this);
return str;
}
template <typename Result>
void to_json(nlohmann::json &j, const WebsocketApiResponse<Result> &req) {
j["id"] = req.id;
j["status"] = req.status;
j["result"] = req.result;
if (!req.error.is_null() || req.status >= 400) {
j["error"] = req.error;
}
}
template <typename Result>
void from_json(const nlohmann::json &j, WebsocketApiResponse<Result> &req) {
j.at("id").get_to(req.id);
j.at("status").get_to(req.status);
auto it = j.find("result");
if (it != j.end()) {
it->get_to(req.result);
}
it = j.find("error");
if (it != j.end()) {
it->get_to(req.error);
}
}
#define RESPONSE_DEF(T) \
template struct WebsocketApiResponse<T>; \
template void to_json( \
nlohmann::json &j, const WebsocketApiResponse<T> &req); \
template void from_json( \
const nlohmann::json &j, WebsocketApiResponse<T> &req)
RESPONSE_DEF(nlohmann::json);
} // namespace binance_execution
@@ -0,0 +1,35 @@
#pragma once
#include "lib/core/FMT_IMPL.h"
#include "lib/io/json_formatter.h"
#include <nlohmann/json.hpp>
namespace binance_execution {
template <typename Result>
struct WebsocketApiResponse {
std::string id;
std::int64_t status;
Result result;
nlohmann::json error;
std::string to_string() const;
};
template <typename Result>
void to_json(nlohmann::json &j, const WebsocketApiResponse<Result> &req);
template <typename Result>
void from_json(const nlohmann::json &j, WebsocketApiResponse<Result> &req);
} // namespace binance_execution
template <typename Result>
FMT_IMPL_T(
binance_execution::WebsocketApiResponse<Result>,
r,
R"({{"id": "{}", "status": {}, "result": {} }})",
r.id,
r.status,
bts::io::json_formatter(r.result));
@@ -112,7 +112,7 @@ OkxExeCfg {
secret_key: "240038D7A28D7F9461247CFCB8E43710"
api_key: "dc188532-c74c-4d40-affe-3c6bf4d0ca9a"
passphrase: ":,7i%tA%ivfj%SCbX?Ut"
ws_url: "wss://wsaws.okx.com:8443"
ws_url: "wss://ws.okx.com:8443"
rest_url: "https://www.okx.com"
ws_ping_timeout_seconds: 10
ws_ping_interval_seconds: 25
@@ -112,7 +112,7 @@ OkxExeCfg {
secret_key: "240038D7A28D7F9461247CFCB8E43710"
api_key: "dc188532-c74c-4d40-affe-3c6bf4d0ca9a"
passphrase: ":,7i%tA%ivfj%SCbX?Ut"
ws_url: "wss://wsaws.okx.com:8443"
ws_url: "wss://ws.okx.com:8443"
rest_url: "https://www.okx.com"
ws_ping_timeout_seconds: 10
ws_ping_interval_seconds: 10
@@ -43,10 +43,6 @@ Logging {
}
file {
default_level: "Info"
subsystem {
name: "md"
level: "Diag"
}
subsystem {
name: "OrderMan"
level: "Diag"