Add udp.py to tools

made the 'toward' behavior for lean an enum

.gitignore

@@ -22,4 +22,5 @@ node_modules
 paket-files
 project.lock.json
 src/server/sparkd/orderman/perf_analysis/data
+src/server/sparkd/orderman/perf_analysis/logs/
 src/server/sparkd/orderman/perf_analysis/trace_data/

src/client/spark/Orders/OrderActivitySummarizer.cs

@@ -134,7 +134,9 @@ namespace bts.spark {
             }
          }
 
-         foreach (var (neworder, oldorder) in update.changes) {
+         foreach (var order in update.changes) {
+            var neworder = order.New;
+            var oldorder = order.Old;
             (var security, var spec) = get_security(neworder.Def.SecurityKey);
             //
             // description

src/client/spark/Orders/OrderTypeVMs/LeanOrderVM.cs

@@ -24,13 +24,15 @@ namespace mbts.spark
             this.child_order_types = new InlineUserProperty<IReadOnlyCollection<CustomOneofCase>>(child_order_types, true);
             cross_threshold = new InlineUserProperty<bts.spark.QuantityMeasure>(
                               new bts.spark.QuantityMeasure() { Ratio = 1.0, Absolute = 1}, false);
-            max_chase_ticks = new InlineUserProperty<uint>(10, false);
+            max_chase_ticks = new InlineUserProperty<uint>(2, false);
+            trade_toward_behavior = new InlineUserProperty<LeanOrderDefinition.Types.OnToward>(LeanOrderDefinition.Types.OnToward.Cancel, false);
          }
          public readonly InlineUserProperty<CustomOneofCase> selected_child_order_type;
          public readonly InlineUserProperty<IReadOnlyCollection<CustomOneofCase>> child_order_types;
          
          public readonly InlineUserProperty<bts.spark.QuantityMeasure> cross_threshold;
          public readonly InlineUserProperty<uint> max_chase_ticks;
+         public readonly InlineUserProperty<LeanOrderDefinition.Types.OnToward> trade_toward_behavior;
       }
 
       class LeanOrderVM : OrderTypeVMBase<LeanOrderDefinition>, IHierarchicalOrderTypeVM, IDisposable 
@@ -44,7 +46,8 @@ namespace mbts.spark
          IDisposable _dispose, _child_order_errors_sub;
          
          public bts.spark.QuantityMeasure cross_threshold { get; set; } = new bts.spark.QuantityMeasure() { Ratio = 1.0, Absolute = 1 };
-         public uint max_chase_ticks { get; set; } = 10;
+         public uint max_chase_ticks { get; set; } = 2;
+         public LeanOrderDefinition.Types.OnToward trade_toward_behavior { get; set; } = LeanOrderDefinition.Types.OnToward.Cancel;      
 
          public LeanOrderVM(OrdersModel model, IObservable<SecurityMD> security_md_source)
             : this(model, security_md_source, security_md_source.Select(smd =>
@@ -77,6 +80,11 @@ namespace mbts.spark
                      max_chase_ticks = p.max_chase_ticks.value;
                      RaisePropertyChanged(nameof(max_chase_ticks));
                   }
+                  if (p.trade_toward_behavior.force_override)
+                  {
+                     trade_toward_behavior = p.trade_toward_behavior.value;
+                     RaisePropertyChanged(nameof(trade_toward_behavior));
+                  }
                   _inline_props = p;
                   _validate();
                })
@@ -94,10 +102,13 @@ namespace mbts.spark
                var i = factory.lean_order();
                i.CrossThreshold = cross_threshold;
                i.MaxChaseTicks = max_chase_ticks;
+               i.TradeTowardBehavior = trade_toward_behavior;
                child_order.assign_to(i.Child);
                value.Lean = i;
             }
-         }         
+         }
+
+         public IEnumerable<LeanOrderDefinition.Types.OnToward> trade_toward_types => EnumHelper.get_values_except<LeanOrderDefinition.Types.OnToward>();         
 
          public ICollection<IOrderTypeVM> child_order_types 
          {

src/client/spark/view/Controls/Order/OrderDetailsPopupControl.xaml

@@ -222,6 +222,8 @@
                           <RowDefinition Height="4" />
                           <RowDefinition Height="Auto"/>
                           <RowDefinition Height="4" />
+                          <RowDefinition Height="Auto"/>
+                          <RowDefinition Height="4" />                          
                           <RowDefinition/>
                         </Grid.RowDefinitions>
                         <Grid.ColumnDefinitions>
@@ -233,9 +235,11 @@
                         <uag:QuantityMeasureControl Grid.Row="1" Grid.Column="2" QuantityMeasure="{Binding cross_threshold}"/>
                         <TextBlock Grid.Row="3" VerticalAlignment="Center">Max Chase Ticks</TextBlock>
                         <c:TickUpDownUInt Grid.Row="3" Grid.Column="2" Value="{Binding max_chase_ticks}"/>
-                        <TextBlock Grid.Row="5">Child</TextBlock>
+                        <TextBlock Grid.Row="5">On Toward</TextBlock>
+                        <ComboBox Grid.Row="5" Grid.Column="2" ItemsSource="{Binding trade_toward_types}" ItemTemplate="{StaticResource EnumDescriptionTemplate}" SelectedItem="{Binding trade_toward_behavior}"/>                        
+                        <TextBlock Grid.Row="7">Child</TextBlock>
                         <uag:OrderDetailsPopupControl
-                          Grid.Row="5" Grid.Column="2"
+                          Grid.Row="7" Grid.Column="2"
                           IsOpen="{Binding child_order.editable_properties_open, FallbackValue=False}"
                           ItemsSource="{Binding child_order_types}"
                           SelectedItem="{Binding child_order}"

src/client/spark/view/Shell/Workspaces/SettingsManagerTool.cs

@@ -0,0 +1,37 @@
+using bts.api.core;
+using bts.api.persistence;
+using bts.api.user;
+using bts.test;
+using mbts.persist;
+using mbts.ui.shell;
+using System;
+using System.Linq;
+
+namespace bts.settings {
+   public class SettingsManagerTool {
+      public static void run() {
+         var sc = new ServiceContext();
+         var users = sc.locate<UserService>();
+         var persist = sc.locate<PersistenceService>("settings");
+         var user_list = users.get_users(new GetUsersRequest());
+         var user_names = user_list.Users.Select(u => u.Login);
+         var from = ConsoleEx.prompt("From User", user_names);
+         var from_persist = new UserUISettings(user_list.Users.First(u => u.Login == from).Id, persist, null);
+
+         var workspaces = from_persist.query<WorkspaceDefinition>(WorkspacesModel.WORKSPACE_NAMESPACE, ".*").Result;
+         var selected_ws = ConsoleEx.prompt("Choose workspace:", workspaces.Select(w => w.key));
+
+         var to = ConsoleEx.prompt("To User", user_names);
+
+         var name = ConsoleEx.input_prompt("Rename workspace?", selected_ws);
+
+         var confirm = ConsoleEx.prompt($"copy workspace '{selected_ws}' from {from} to {to} and name it '{name}'. Confirm?", "Yes", "No");
+         if (confirm == "Yes") {
+            var ws_blob = workspaces.First(w => w.key == selected_ws).value;
+            var to_persist = new UserUISettings(user_list.Users.First(u => u.Login == to).Id, persist, null);
+            to_persist.set(WorkspacesModel.WORKSPACE_NAMESPACE, name, ws_blob).Wait();
+            Console.WriteLine("Done");
+         }
+      }
+   }
+}

src/client/spark_api/api/Update.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Collections.Immutable;
@@ -53,28 +54,43 @@ namespace bts.api.core {
 #endif
    }
 
+   public readonly struct Change<T> {
+      public Change(T o, T n) {
+         Old = o;
+         New = n;
+      }
+      public readonly T Old;
+      public readonly T New;
+      public Change<T> merge(Change<T> newer) {
+         return new Change<T>(Old, newer.New);
+      }
+      public override string ToString() {
+         return $"{{{Old}=>{New}}}";
+      }
+   }
+
    public class Update<T, TKey> where TKey : IEquatable<TKey> {
       private Func<T, TKey> _key;
       internal Update(
          IReadOnlyCollection<T> all,
          IEnumerable<T> added,
-         IEnumerable<(T New,T Old)> changes,
+         IEnumerable<Change<T>> changes,
          IEnumerable<T> removed,
          Func<T, TKey> key_extractor) {
          this.added = added ?? new T[0];
          this.removed = removed ?? new T[0];
-         this.changes = changes ?? new (T,T)[0];
+         this.changes = changes ?? new Change<T>[0];
          this.all = all;
          _key = key_extractor;
       }
 
       public static Update<T, TKey> empty(Func<T, TKey> key_extractor) =>
-         new Update<T, TKey>(new T[0], new T[0], new (T,T)[0], new T[0], key_extractor);
+         new Update<T, TKey>(new T[0], new T[0], new Change<T>[0], new T[0], key_extractor);
 
       public Update<T,TKey> merge(Update<T, TKey> next) {
          var comp = new RowEqualityComparer<T, TKey>(_key);
          var added = this.added as HashSet<T> ?? new HashSet<T>(this.added, comp);
-         var changes = this.changed as HashSet<(T New,T Old)> ?? new HashSet<(T,T)>(this.changes, comp);
+         var changes = this.changed as ChangeDict<T,TKey> ?? new ChangeDict<T,TKey>(this.changes, _key);
          var removed = this.removed as HashSet<T> ?? new HashSet<T>(this.removed, comp);
 
          // Add
@@ -87,9 +103,7 @@ namespace bts.api.core {
                added.Add(c.New);
             }
             else {
-               // Remove does not error if it was not present
-               changes.Remove(c);
-               changes.Add(c);
+               changes.AddOrMerge(c);
             }
          }
 
@@ -98,7 +112,7 @@ namespace bts.api.core {
             if (!added.Remove(r)) {
                removed.Add(r);
             }
-            changes.Remove((r,r));
+            changes.Remove(_key(r));
          }
 
          return new Update<T, TKey>(
@@ -110,7 +124,7 @@ namespace bts.api.core {
 
       public IReadOnlyCollection<T> all { get; }
       public IEnumerable<T> added { get; }
-      public IEnumerable<(T New,T Old)> changes { get; }
+      public IEnumerable<Change<T>> changes { get; }
       public IEnumerable<T> changed => changes.Select(c => c.New);
       public IEnumerable<T> removed { get; }
 
@@ -127,7 +141,40 @@ namespace bts.api.core {
 #endif
    }
 
-   struct RowEqualityComparer<T,TKey> : IEqualityComparer<T>, IEqualityComparer<(T,T)>
+   class ChangeDict<T, TKey> : IEnumerable<Change<T>> {
+      Dictionary<TKey, Change<T>> _changes;
+      Func<Change<T>, TKey> _key;
+
+      public ChangeDict(IEnumerable<Change<T>> changes, Func<T, TKey> key) {
+         _key = x => key(x.New);
+         _changes = changes.ToDictionary(_key);
+      }
+
+      public Change<T>? AddOrMerge(Change<T> newer) {
+         var key = _key(newer);
+         if (_changes.TryGetValue(key, out Change<T> old)) {
+            _changes[key] = old.merge(newer);
+            return old;
+         }
+         _changes.Add(key, newer);
+         return null;
+      }
+
+      public void Remove(TKey key) {
+         _changes.Remove(key);
+      }
+
+      #region IEnumerable
+      public IEnumerator<Change<T>> GetEnumerator() {
+         return _changes.Values.GetEnumerator();
+      }
+      IEnumerator IEnumerable.GetEnumerator() {
+         return _changes.Values.GetEnumerator();
+      }
+      #endregion
+   }
+
+   struct RowEqualityComparer<T,TKey> : IEqualityComparer<T>, IEqualityComparer<Change<T>>
       where TKey : IEquatable<TKey> {
       readonly Func<T,TKey> _key;
 
@@ -139,16 +186,16 @@ namespace bts.api.core {
          return _key(x).Equals(_key(y));
       }
 
-      public bool Equals((T, T) x, (T, T) y) {
-         return Equals(x.Item1, y.Item1);
+      public bool Equals(Change<T> x, Change<T> y) {
+         return Equals(x.New, y.New);
       }
 
       public int GetHashCode(T obj) {
          return _key(obj).GetHashCode();
       }
 
-      public int GetHashCode((T, T) obj) {
-         return GetHashCode(obj.Item1);
+      public int GetHashCode(Change<T> c) {
+         return GetHashCode(c.New);
       }
    }
 
@@ -159,7 +206,7 @@ namespace bts.api.core {
       internal TableUpdate(
          ImmutableDictionary<TKey, T> init,
          IEnumerable<T> added,
-         IEnumerable<(T,T)> changed,
+         IEnumerable<Change<T>> changed,
          IEnumerable<T> removed) {
          _state = init;
          this.added = added;
@@ -169,7 +216,7 @@ namespace bts.api.core {
 
       public IReadOnlyDictionary<TKey, T> state { get { return _state; } }
       public IEnumerable<T> added { get; }
-      public IEnumerable<(T New,T Old)> changes { get; }
+      public IEnumerable<Change<T>> changes { get; }
       public IEnumerable<T> changed => changes.Select(c => c.New);
 
       public IEnumerable<T> removed { get; }
@@ -185,7 +232,7 @@ namespace bts.api.core {
       }
 
       internal TableUpdate<T, TKey> as_snapshot() {
-         return new TableUpdate<T, TKey>(_state, _state.Values.ToList(), new (T,T)[0], new T[0]);
+         return new TableUpdate<T, TKey>(_state, _state.Values.ToList(), new Change<T>[0], new T[0]);
       }
    }
 
@@ -208,7 +255,7 @@ namespace bts.api.core {
             if (raw.is_reset) {
                var removed = state.Values.ToList();
                state.Clear();
-               return new Update<T, TKey>(new T[0], new T[0], new (T,T)[0], removed, key_extractor);
+               return new Update<T, TKey>(new T[0], new T[0], new Change<T>[0], removed, key_extractor);
             }
             else {
                var removed = new List<T>(raw.removed_keys.Count);
@@ -222,11 +269,11 @@ namespace bts.api.core {
                for (int x = 0; x < raw.added_items.Count; ++x) {
                   state[raw.added_keys[x]] = raw.added_items[x];
                }
-               List<(T, T)> changes = new List<(T, T)>(raw.changed_items.Count);
+               var changes = new List<Change<T>>(raw.changed_items.Count);
                for (int x = 0; x < raw.changed_items.Count; ++x) {
                   var old = state[raw.changed_keys[x]];
                   state[raw.changed_keys[x]] = raw.changed_items[x];
-                  changes.Add((raw.changed_items[x], old));
+                  changes.Add(new Change<T>(old, raw.changed_items[x]));
                }
                return new Update<T, TKey>(state.Values.ToList(), raw.added_items.ToList(), changes, removed, key_extractor);
             }
@@ -235,7 +282,7 @@ namespace bts.api.core {
 
       public static IObservable<TableUpdate<T,TKey>> Table<T,TKey>(this IObservable<Update<T,TKey>> source) where TKey : IEquatable<TKey> =>
          source.Scan(
-            new TableUpdate<T,TKey>(ImmutableDictionary<TKey,T>.Empty, new T[0], new (T,T)[0], new T[0]),
+            new TableUpdate<T,TKey>(ImmutableDictionary<TKey,T>.Empty, new T[0], new Change<T>[0], new T[0]),
             (table, update) => TableUpdate<T,TKey>.update(table,update));
 
       // Fold updates and emit at max 1/interval

src/client/spark_core/TestRunner/ConsoleEx.cs

@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 
 namespace bts.test {
    public static class ConsoleEx {
@@ -21,17 +23,34 @@ namespace bts.test {
       }
 
       public static T prompt<T>(string prompt, params T[] options) {
-         Console.WriteLine(prompt);
-         for (int x = 1; x <= options.Length; ++x) {
+         return ConsoleEx.prompt(prompt, (IEnumerable<T>)options);
+      }
+      public static T prompt<T>(string prompt, IEnumerable<T> options) {
+         WriteLine(prompt, ConsoleColor.Yellow);
+         for (int x = 0; x < options.Count(); ++x) {
             Write($"{x}. ", ConsoleColor.Yellow);
-            WriteLine(options[x].ToString());
+            WriteLine(options.ElementAt(x).ToString());
          }
-         int response;
-         do {
-            Int32.TryParse(Console.ReadKey().ToString(), out response);
-            if (response > options.Length) response = 0;
-         } while (response == 0);
-         return options[response];
+         int response = 0;
+         bool success = false;
+         while(!success) {
+            var key = Console.ReadKey();
+            success = Int32.TryParse(key.KeyChar.ToString(), out response);
+            if (response > options.Count()) success = false;
+         }
+         var value = options.ElementAt(response);
+         Console.WriteLine(" >> " + value.ToString());
+         Console.WriteLine();
+         return value;
+      }
+
+      public static string input_prompt(string prompt, string default_result) {
+         Write(prompt + " (", ConsoleColor.Yellow);
+         Write(default_result, ConsoleColor.White);
+         Write(") ", ConsoleColor.Yellow);
+         var result = Console.ReadLine();
+         if (string.IsNullOrWhiteSpace(result)) return default_result;
+         return result;
       }
 
 

src/client/spark_test/api/ApiUpdateTests.cs

@@ -227,5 +227,15 @@ namespace mbts.api.Test {
          await validate_naggle(raw);
       }
 
+      [Test]
+      public void test_merge_begin_end_state() {
+         int key(string s) => s[0];
+         var empty = new string[0];
+         var a2 = new Update<string, int>(new[] { "a2" }, empty, new[] {  new Change<string>("a1", "a2") }, empty, key);
+         var a3 = new Update<string, int>(new[] { "a3" }, empty, new[] { new Change<string>("a2", "a3") }, empty, key);
+         var merged = a2.merge(a3);
+         Assert.SequenceEqual(new[] { new Change<string>("a1", "a3") }, merged.changes);
+      }
+
    }
 }

src/client/sparkcmd/Cmd/SettingsManagerCommand.cs

@@ -0,0 +1,20 @@
+using ManyConsole;
+using System.Reflection;
+
+namespace bts {
+
+   public class SettinsManagerCommand : ConsoleCommand {
+      public SettinsManagerCommand() {
+         IsCommand("SettingsManager", "Copy workspaces around");
+      }
+
+      public override int Run(string[] remainingArguments) {
+         var asm = Assembly.Load("spark");
+         var type = asm.GetType("bts.settings.SettingsManagerTool");
+         var run = type.GetMethod("run", BindingFlags.Static | BindingFlags.Public);
+         run.Invoke(null, new object[0]);
+         return 0;
+      }
+   }
+
+}

src/lib/CMakeDeps.cmake

@@ -1 +1 @@
-set(sources api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/services/AdminService.proto;api/services/ApiUnitTestService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MarketDataRelay.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/services/UserService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/Executor.conf;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cpp_config.conf;config/defaults/dynamic_orders.conf;config/defaults/edge.conf;config/defaults/expiration_display_offset.conf;config/defaults/jetfix.conf;config/defaults/logging.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/ExchangeSimCfg.proto;config/protos/MarketData.proto;config/protos/Ports.proto;config/protos/Symbols.proto;config/protos/UI.proto;config/protos/cpp_config.proto;config/protos/dynamic_orders.proto;config/protos/environment.proto;config/protos/expiration_display_offset.proto;config/protos/jetfix.proto;config/protos/logging.proto;config/protos/orders_fills_cleanup.proto;config/protos/quickbus_cfg.proto;config/protos/spreader.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/toggles.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPriceLantencyTester.cpp;md/Diag/ShmDumper.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MDRecoverableConsumer.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/md2_tests.cpp;md/test/synthetic_book_merge_tests.cpp;os/os.cpp;os/tests/os_tests.cpp;quickbus/core/LocalNode.cpp;quickbus/core/Quickbus.cpp;quickbus/core/SubscriptionID.cpp;quickbus/core/TableID.cpp;quickbus/message_model/Update.cpp;quickbus/message_model/merge_updates.cpp;quickbus/network/QBConnection.cpp;quickbus/network/QBConnectionManager.cpp;quickbus/network/endpoint_description.cpp;quickbus/network/quickbus_msgs.proto;quickbus/pubsub/TablePublisher.cpp;quickbus/pubsub/TableSubscriber.cpp;quickbus/pubsub/tests/qb_pubsub_tests.cpp;quickbus/routing/INode.cpp;quickbus/routing/NodeID.cpp;quickbus/routing/Router.cpp;quickbus/rpc/RPCClient.cpp;quickbus/rpc/RPCServer.cpp;quickbus/rpc/RPCSessionID.cpp;quickbus/rpc/rpc_utils.cpp;quickbus/rpc/tests/qb_rpc_tests.cpp;quickbus/tests/quickbus_tests.cpp;sparkproto/SparkProtos.proto;testing/catch_test_runner.cpp;utils/Dispatcher.cpp;utils/HTTPStream.cpp;utils/PBUtils.cpp;utils/Puid.cpp;utils/SecureLogon.cpp;utils/SharedMemory.cpp;utils/Sqlite_utils.cpp;utils/Uuid.cpp;utils/datetime.cpp;utils/download.cpp;utils/file_utils.cpp;utils/init_spark_lib.cpp;utils/is_unittest_environment.cpp)
+set(sources api/apicore.proto;api/apigateway.proto;api/apioptions.proto;api/core/ApiClientInterop.cpp;api/core/ApiServerInterop.cpp;api/public/ApiServer.cpp;api/public/ClientChannel.cpp;api/public/QuickbusClientChannel.cpp;api/public/ServerChannel.cpp;api/public/ServiceContext.cpp;api/service_helpers/AdminServiceUtils.cpp;api/services/AdminService.proto;api/services/ApiUnitTestService.proto;api/services/ErrorReportService.proto;api/services/HelloService.proto;api/services/MarketDataRelay.proto;api/services/OrderService.proto;api/services/OutOfProcessURLListener.proto;api/services/PersistenceService.proto;api/services/PositionService.proto;api/services/ShellDiagnosticService.proto;api/services/ShellLatencyService.proto;api/services/SyntheticSecurityService.proto;api/services/UserService.proto;api/test/ApiTests.cpp;api/test/HelloServiceServer.cpp;config/ConfigCheck.cpp;config/ConfigDownloader.cpp;config/ConfigErrorCollector.cpp;config/ConfigLog.cpp;config/ConfigPropertyEvaluator.cpp;config/ConfigValueQueue.cpp;config/ConfigurationException.cpp;config/ConfigurationState.cpp;config/PreprocessInputStream.cpp;config/config.cpp;config/defaults/Executor.conf;config/defaults/MarketData.conf;config/defaults/Ports.conf;config/defaults/Symbols.conf;config/defaults/cpp_config.conf;config/defaults/dynamic_orders.conf;config/defaults/edge.conf;config/defaults/expiration_display_offset.conf;config/defaults/jetfix.conf;config/defaults/logging.conf;config/defaults/orders_fills_cleanup.conf;config/defaults/spreader.conf;config/defaults/testkeys.conf;config/defaults/toggles.conf;config/extensions/extensions.proto;config/invalid/invalid.proto;config/protos/ExchangeSimCfg.proto;config/protos/MarketData.proto;config/protos/Ports.proto;config/protos/Symbols.proto;config/protos/UI.proto;config/protos/cpp_config.proto;config/protos/dynamic_orders.proto;config/protos/expiration_display_offset.proto;config/protos/jetfix.proto;config/protos/logging.proto;config/protos/orders_fills_cleanup.proto;config/protos/quickbus_cfg.proto;config/protos/spreader.proto;config/protos/system.proto;config/protos/testkeys.proto;config/protos/toggles.proto;config/tests/HTTPServer.cpp;config/tests/config_ref_tests.cpp;config/tests/config_tests.cpp;config/text_format/text_format.cpp;domain_model/Security.cpp;domain_model/Security.proto;domain_model/SecurityTests.cpp;domain_model/tests/domain_tests.cpp;expression/BooleanGrammar.cpp;expression/ExpressionEvaluator.cpp;expression/ParserExceptions.cpp;expression/Value.cpp;expression/tests/expression_parser_tests.cpp;logging/LoggerHub.cpp;logging/Sink.cpp;logging/logging.cpp;logging/tests/logging_tests.cpp;md/Diag/MDPriceLantencyTester.cpp;md/Diag/ShmDumper.cpp;md/MDConsumer.cpp;md/MDConsumerHandle.cpp;md/MDConsumerInterop.cpp;md/MDProducer.cpp;md/MDProducerHandle.cpp;md/MDRecoverableConsumer.cpp;md/MemorySegmentUtils.cpp;md/PlatformLoader.cpp;md/PriceSize.cpp;md/SecurityDB.cpp;md/SecurityIDRecorder.cpp;md/SettlementRecorder.cpp;md/Shm/FixedSecurity.cpp;md/SyntheticSecurityService.cpp;md/cme/MDP3Utils.cpp;md/common.cpp;md/protos/co2.proto;md/relay/QueryServer.cpp;md/relay/RelayClient.cpp;md/relay/RelayRecorder.cpp;md/test/Co2ManagedTestHelper.cpp;md/test/Co2TestInstrument.cpp;md/test/md2_tests.cpp;md/test/synthetic_book_merge_tests.cpp;os/os.cpp;os/tests/os_tests.cpp;quickbus/core/LocalNode.cpp;quickbus/core/Quickbus.cpp;quickbus/core/SubscriptionID.cpp;quickbus/core/TableID.cpp;quickbus/message_model/Update.cpp;quickbus/message_model/merge_updates.cpp;quickbus/network/QBConnection.cpp;quickbus/network/QBConnectionManager.cpp;quickbus/network/endpoint_description.cpp;quickbus/network/quickbus_msgs.proto;quickbus/pubsub/TablePublisher.cpp;quickbus/pubsub/TableSubscriber.cpp;quickbus/pubsub/tests/qb_pubsub_tests.cpp;quickbus/routing/INode.cpp;quickbus/routing/NodeID.cpp;quickbus/routing/Router.cpp;quickbus/rpc/RPCClient.cpp;quickbus/rpc/RPCServer.cpp;quickbus/rpc/RPCSessionID.cpp;quickbus/rpc/rpc_utils.cpp;quickbus/rpc/tests/qb_rpc_tests.cpp;quickbus/tests/quickbus_tests.cpp;sparkproto/SparkProtos.proto;testing/catch_test_runner.cpp;utils/Dispatcher.cpp;utils/HTTPStream.cpp;utils/PBUtils.cpp;utils/Puid.cpp;utils/SecureLogon.cpp;utils/SharedMemory.cpp;utils/Sqlite_utils.cpp;utils/Uuid.cpp;utils/datetime.cpp;utils/download.cpp;utils/file_utils.cpp;utils/init_spark_lib.cpp;utils/is_unittest_environment.cpp)

src/lib/api/services/OrderService.proto

@@ -109,6 +109,15 @@ message TickOrderDefinition {
 message LeanOrderDefinition {
   spark.QuantityMeasure cross_threshold = 1;
   uint32 max_chase_ticks = 2;
+  reserved 4; reserved "detach_on_toward";
+  enum OnToward
+  {
+     NO_CHANGE = 0; //continue to lean on initial inside market
+     CANCEL = 1; //cancel algo and order
+     DETACH = 2; //cancel algo but leave order working
+     REPRICE = 3; //lean on new inside market
+  }
+  OnToward trade_toward_behavior = 5;
   CustomOrderDefinition child = 3;
 }
 message JoinOrderDefinition {

src/lib/config/ConfigPropertyEvaluator.cpp

@@ -234,7 +234,7 @@ bool ConfigPropertyEvaluator::get_value(const std::string& property, Value* out_
    tokenizer<char_separator<char> >::iterator iter = tokens.begin (), end = tokens.end ();
 
    // The first token is the key name
-   const proto::Message* key = _config.try_get_message(*iter);
+   const proto::Message* key = _config.try_get_message_or_default(*iter);
    if (!key) return false;
 
    // 2) Drill down using reflection to find the named property

src/lib/config/ConfigurationState.cpp

@@ -616,6 +616,19 @@ const proto::Message* ConfigurationState::try_get_message(const std::string& pro
    return nullptr;
 }
 
+const proto::Message* ConfigurationState::try_get_message_or_default(const std::string& property_name) {
+   const auto pmsg = try_get_message(property_name);
+   if (pmsg) {
+      return pmsg;
+   }
+   // If we can't find a value we should return a default instance
+   const auto* descriptor = get_descriptor_by_name(property_name);
+   if (descriptor) {
+      return proto::MessageFactory::generated_factory()->GetPrototype(descriptor);
+   }
+   return nullptr;
+}
+
 const proto::Message& ConfigurationState::get_message(const proto::Descriptor* property_desc)
 {
    if (!is_valid_key_descriptor(property_desc))

src/lib/config/ConfigurationState.h

@@ -59,6 +59,7 @@ namespace internal {
 
       const proto::Message& get_message(const proto::Descriptor* property_desc);
       const proto::Message* try_get_message(const std::string& property_name);
+      const proto::Message* try_get_message_or_default(const std::string& property_name);
       
       bool is_initialized() { return _is_initialized; }
 

src/lib/config/config.cpp

@@ -76,16 +76,7 @@ extern "C" BTS_EXPORT const char* get_buffer(const char* key_name, int& out_leng
 
 extern "C" BTS_EXPORT const proto::Message* get_named(const char* key_name)
 {
-   const auto* pmessage = config_state.try_get_message(key_name);
-   if (pmessage) {
-      return pmessage;
-   }
-   // If we can't find a value we should return a default instance
-   const auto* descriptor = config_state.get_descriptor_by_name(key_name);
-   if (descriptor) {
-      return proto::MessageFactory::generated_factory()->GetPrototype(descriptor);
-   }
-   return nullptr;
+   return config_state.try_get_message_or_default(key_name);
 }
 
 BTS_EXPORT std::string get_macro_string(const char* name) { return config_state.get_macro_string(name); }

src/lib/config/defaults/Symbols.conf

@@ -26,6 +26,26 @@ SymbolList {
     }
 }
 
+Symbols {
+    rule: "DeploymentType.type == 'DEMO' and System.platform == 'windows'"
+    value {
+        sim {
+            future: @SymbolList.symbol
+            spread: @SymbolList.symbol
+        }
+    }
+}
+
+Symbols {
+    rule: "DeploymentType.type == 'PROD' and System.platform == 'windows'"
+    value {
+        glbx {
+            future: @SymbolList.symbol
+            spread: @SymbolList.symbol
+        }
+    }
+}
+
 Symbols {
     value {
         glbx {

src/lib/config/protos/environment.proto

@@ -1,8 +0,0 @@
-syntax='proto2';
-package bts.config;
-
-message Deployment {
-  // Acceptable values:
-  // DEV_CHAPEL_HILL, DEV_CHICAGO, DEMO_GT, QA, CUSTOMER
-  optional string type = 1 [default="DEV_CHAPEL_HILL"];  
-}

src/lib/jetfix/connection/SocketConnection.cpp

@@ -132,7 +132,7 @@ int SocketConnection::write(WriteVec const &vec, size_t vec_size,
           "perf_event:tick_to_order s2s_us:{} fix_socket_r:{} md:{} "
           "md_src:{} spreader_on_book:{} spreader_modify_order:{} "
           "spreader_on_child_order_update:{} " 
-          "order_update:{} exe_gw:{} fix_socket_w:{} md_sn:{}",
+          "order_update:{} exe_gw:{} fix_socket_w:{} md_sn:{} order_key:{}",
           s2s.microseconds(), trace_cpy.fix_socket_r.to_uint64(),
           trace_cpy.md.to_uint64(), trace_cpy.md_src.to_uint64(),
           trace_cpy.spreader_on_book.to_uint64(),
@@ -140,7 +140,8 @@ int SocketConnection::write(WriteVec const &vec, size_t vec_size,
           trace_cpy.spreader_on_child_order_update.to_uint64(), 
           trace_cpy.order_update.to_uint64(),
           trace_cpy.exe_gw.to_uint64(), trace_cpy.fix_socket_w.to_uint64(),
-          trace_cpy.md_sn);
+          trace_cpy.md_sn,
+          trace_cpy.order_key);
       _tick_to_order->observe(s2s.nanoseconds());
    }
 
@@ -150,7 +151,7 @@ int SocketConnection::write(WriteVec const &vec, size_t vec_size,
           "perf_event:fill_to_order s2s_us:{} fix_socket_r:{} md:{} md_src:{} "
           "spreader_on_book:{} spreader_modify_order:{} "
           "spreader_on_child_order_update:{} " 
-          "order_update:{} exe_gw:{} fix_socket_w:{} md_sn:{}",
+          "order_update:{} exe_gw:{} fix_socket_w:{} md_sn:{} order_key:{}",
           s2s.microseconds(), trace_cpy.fix_socket_r.to_uint64(),
           trace_cpy.md.to_uint64(), trace_cpy.md_src.to_uint64(),
           trace_cpy.spreader_on_book.to_uint64(),
@@ -158,7 +159,8 @@ int SocketConnection::write(WriteVec const &vec, size_t vec_size,
           trace_cpy.spreader_on_child_order_update.to_uint64(), 
           trace_cpy.order_update.to_uint64(),
           trace_cpy.exe_gw.to_uint64(), trace_cpy.fix_socket_w.to_uint64(),
-          trace_cpy.md_sn);
+          trace_cpy.md_sn,
+          trace.order_key);
       _fill_to_order->observe(s2s.nanoseconds());
    }
 

src/lib/md/Diag/ShmDumper.cpp

@@ -76,8 +76,7 @@ string dump_security(const FixedSecurity& security) {
    res << "'main_frac': " << (int)security.cme_main_fraction << ", ";
    res << "'sub_fract': " << (int)security.cme_sub_fraction << ", ";
    res << "'price_format': " << (int)security.cme_price_display_format << ", ";
-
-   res << "'market_segment': " << security.cme_market_segment_id << ", ";
+   res << "'market_segment': " << (int)security.cme_market_segment_id << ", ";
 
    res << "'price_break': " << security.price_break << ", ";
    res << "'low_tick': " << security.low_tick << ", ";

src/lib/orderman/actors/OrderStore.cpp

@@ -24,8 +24,7 @@ struct OrderStore::PImpl {
    PImpl()
        : state(State::WAIT_SEC), known_orders(),
          store(new persistence::SqliteStore("orders")), order_stream(),
-         soft_cleanup_enabled (config::get<config::toggles::OrdersSoftCleanup>().enabled()),
-         cutoff_time(SODPosition::get_orders_start_time())
+         soft_cleanup_enabled (config::get<config::toggles::OrdersSoftCleanup>().enabled())
          {}
 
    State state;
@@ -76,6 +75,8 @@ void OrderStore::init() {
             auto all = me->store->retrieve_all()[orders_ns];
             LOG(persist, Info, "Finished loading all {} orders", all.size());
 
+            me->cutoff_time=SODPosition::get_order_cutoff_time(all);
+
             for (auto const &kvp : all)
                if (utils::pb::from_string<api::orderman::OrderRecord>(
                        kvp.second, &rec)) {

src/lib/orderman/actors/SODPosition.cpp

@@ -4,6 +4,7 @@
 #include "lib/config/protos/toggles.pb.h"
 #include "lib/config/protos/orders_fills_cleanup.pb.h"
 #include "lib/config/config.h"
+#include "lib/utils/PBUtils.h"
 #include <sys/time.h>
 #include <sstream>
 
@@ -18,12 +19,14 @@ _sod_time(_get_fills_sod_time()){
                       _enabled, _sod_time.tv_sec);
 }
 
+// returns the timeval for today's hh,mm,ss
+// timeval is the microsecons from epoch)
 timeval SODPosition::_to_timeval(uint16_t hh, uint16_t mm, uint16_t ss){
    bts_assert(hh<24);
    bts_assert(mm<60);
    bts_assert(ss<60);
    time_t secs;
-   time (&secs);  // seconds since epoch
+   time (&secs);  // seconds since epoch (utc)
    struct tm* ptm = gmtime (&secs);
    ptm->tm_hour=hh;
    ptm->tm_min=mm;
@@ -37,9 +40,8 @@ timeval SODPosition::_to_timeval(uint16_t hh, uint16_t mm, uint16_t ss){
 
 timeval SODPosition::_subtract_days(const timeval& tv, uint32_t n_days){
    uint32_t secs_in_day=86400*n_days;
-   bts_assert(tv.tv_sec>secs_in_day);
    timeval rv=tv;
-   rv.tv_sec=rv.tv_sec-secs_in_day;
+   rv.tv_sec=(rv.tv_sec > secs_in_day) ? rv.tv_sec-secs_in_day : 0;
    return rv;
 }
 
@@ -69,6 +71,42 @@ timeval SODPosition::get_orders_start_time(){
    return sod;
 }
 
+timeval SODPosition::get_order_cutoff_time(
+                        const std::map<std::string,std::string>& order_map){
+   //returns true is t1 >= t2
+   auto compare_ts = [&](google::protobuf::Timestamp t1, google::protobuf::Timestamp t2) -> bool
+   {
+      if (t1.seconds() > t2.seconds())
+         return true;
+      else if (t1.seconds() < t2.seconds())
+         return false;
+      else
+         return t1.nanos() >= t2.nanos();
+   };                        
+   google::protobuf::Timestamp max_ts;
+   max_ts.set_seconds(0);
+   max_ts.set_nanos(0);
+
+   for (auto const &kvp : order_map){
+      OrderRecord trec;
+      if (utils::pb::from_string<api::orderman::OrderRecord>(kvp.second, &trec)){
+         const auto& ts = trec.state().created();
+         if(compare_ts(ts,max_ts))
+            (&max_ts)->CopyFrom(ts);
+      }                  
+      else
+         LOG(OrderMan, Error,
+             "Could not recover order {} from storage. "
+             "This should be investigated",
+             kvp.first);
+   }
+   timeval rv;
+   rv.tv_sec=max_ts.seconds();
+   rv.tv_usec=static_cast<double>(max_ts.nanos())/1000;
+   const auto& orders_cleanup = config::get<config::OrdersSoftCleanupCutoff>();
+   return _subtract_days(rv,orders_cleanup.extra_days());
+}
+
 std::string SODPosition::_sod_fill_id(const Fill& fill,api::Side side){
    //sod:acct:long/short:sec
    std::ostringstream oss;
@@ -80,12 +118,12 @@ std::string SODPosition::_sod_fill_id(const Fill& fill,api::Side side){
 void SODPosition::_merge_pos(Fill& pos, const Fill& fill){
    bts_assert(pos.security_key()==fill.security_key());
    bts_assert(pos.account()==fill.account());
-   bts_assert(pos.side()==fill.side());
-   pos.set_last_price(fill.last_price());
+   bts_assert(pos.side()==fill.side());   
    double total_qty=pos.last_shares()+fill.last_shares();
    double avg_price=(pos.avg_price()*pos.last_shares()+
-                    fill.avg_price()*fill.last_shares())/total_qty;
+                    fill.last_price()*fill.last_shares())/total_qty;
    pos.set_avg_price(avg_price);
+   pos.set_last_price(avg_price);
    pos.set_last_shares(total_qty);
 }
 
@@ -98,6 +136,7 @@ void SODPosition::_zero_pos(Fill& pos, const Fill& fill, api::Side side){
    pos.set_fill_kind(Fill::FILL_KIND_SOD);
    pos.set_order_key(0);
    pos.set_fill_key(_sod_fill_id(fill,side));
+   pos.set_last_shares(0);
 }
 
 bool SODPosition::add_fill_to_sod_pos(const Fill& fill){

src/lib/orderman/actors/SODPosition.h

@@ -31,9 +31,10 @@ public:
    SODPosition();
    // all inactive orders before this time are not published
    static timeval get_orders_start_time();
+   static timeval get_order_cutoff_time(const std::map<std::string,std::string>& ordermap);
    bool add_fill_to_sod_pos(const Fill& fill);
    void add_sod_pos_as_fill_events(std::vector<StoredFillEvent>& fill_events,
-                        std::map<SecurityKey, const api::Security*> sec_map);
+                        std::map<SecurityKey, const api::Security*> sec_map);   
 };
 
 }}

src/lib/orderman/orderman_utils/TickToTradeCorID.h

@@ -16,6 +16,7 @@ public:
    utils::TimeStamp exe_gw ;
    utils::TimeStamp fix_socket_w ;
    utils::TimeStamp spreader_modify_order ;
+   uint32_t order_key = 0;
    uint16_t md_sn = 0;
 };
 

src/lib/orderman/orders/NativeOrder.cpp

@@ -271,7 +271,10 @@ NativeOrder::~NativeOrder() {}
 
 void NativeOrder::on_activate(Price p, Quantity q,
                               bts::orderman::OrderServices &services,
-                              PerfTrace const&trace) {
+                              PerfTrace const&t) {
+   auto trace = t;
+   trace.order_key = key().value();
+   
    tracepoint(bts_trace, trace_order, (char *)"on_activate", 0, trace.md_sn, 0);
    bts_assert(!me->services);
    me->services = &services;
@@ -454,6 +457,7 @@ void NativeOrder::on_activate(Price p, Quantity q,
 void NativeOrder::on_update(Price p, Quantity q, PerfTrace const &t) {
    auto trace = t;
    trace.order_update = utils::TimeStamp::now();
+   trace.order_key = key().value();
    tracepoint(bts_trace, trace_order, (char *)"on_update", 0, trace.md_sn, 0);
 
    switch (me->state.status()) {

src/lib/orderman/orders/book_algos/BookAlgo.cpp

@@ -408,14 +408,14 @@ void BookAlgo::report_state()
    me->services->report_state(merged_state);
 }
 
-void BookAlgo::payup(const md::PriceSize& opposite, uint32_t cross_ticks)
+void BookAlgo::payup(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks)
 {
    double new_price = opposite.price() + me->tick_size * cross_ticks;
    me->child.update(Price(new_price), Quantity(this_state().quantity()));
    report_state();
 }
 
-void BookAlgo::paydown(const md::PriceSize& opposite, uint32_t cross_ticks)
+void BookAlgo::paydown(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks)
 {
    double new_price = opposite.price() - me->tick_size * cross_ticks;
    me->child.update(Price(new_price), Quantity(this_state().quantity()));

src/lib/orderman/orders/book_algos/BookAlgo.h

@@ -42,8 +42,8 @@ protected:
    virtual void transition(api::orderman::OrderStatus new_state, std::string = "");
    virtual void report_state();    
    
-   virtual void payup(const md::PriceSize&, uint32_t cross_ticks = 0);
-   virtual void paydown(const md::PriceSize&, uint32_t cross_ticks = 0);
+   virtual void payup(const md::PriceSize&, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks = 0);
+   virtual void paydown(const md::PriceSize&, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks = 0);
    void detach(bts::api::orderman::OrderStatus status_after_detach = bts::api::orderman::ORDER_STATUS_CANCELED);       
 private:
    struct PImpl;

src/lib/orderman/orders/book_algos/LeanOrder.cpp

@@ -12,12 +12,15 @@ using namespace api::orderman;
 
 struct LeanOrder::PImpl 
 {
-   PImpl() {}
+   PImpl() : cross_threshold(), max_chase_ticks(0), trade_toward_behavior(LeanOrderDefinition_OnToward_NO_CHANGE), opposite_at_inception(), num_chases(0)
+   {}
    
    spark::QuantityMeasure cross_threshold;    
    uint32_t max_chase_ticks; 
+   LeanOrderDefinition_OnToward trade_toward_behavior;
   
    md::PriceSize opposite_at_inception;   
+   uint32_t num_chases;
 };
 
 
@@ -34,25 +37,40 @@ void LeanOrder::setup_child(const api::orderman::OrderDefinition& this_def, cons
    
    me->cross_threshold = lean_def.cross_threshold();
    me->max_chase_ticks = lean_def.max_chase_ticks();      
+   me->trade_toward_behavior = lean_def.trade_toward_behavior();      
 }
 
-void LeanOrder::payup(const md::PriceSize& opposite, uint32_t cross_ticks)
+void LeanOrder::payup(const md::PriceSize&, const bts::orderman::BookUpdateEvent& book, uint32_t cross_ticks /*= 0*/)
 {
-   double new_price = opposite.price() + tick_size() * cross_ticks;
+   double new_price = me->opposite_at_inception.price() + tick_size() * cross_ticks; 
    double limit = me->opposite_at_inception.price() + tick_size() * me->max_chase_ticks;    
    if(new_price <= limit)
       update_child(Price(new_price), Quantity(this_state().quantity()));
    else 
    {
-      update_child(Price(limit), Quantity(this_state().quantity()));
+      update_child(Price(limit), Quantity(this_state().quantity()));      
       detach();
-   }
+      return;
+   }            
+         
+   const bts::orderman::BookSide& sellbook = book.book.side(api::Side::SIDE_SELL);
+   for(const auto& i : sellbook)
+   {
+      me->opposite_at_inception = i;
+      if(me->opposite_at_inception.price() > new_price)
+         break;
+   }        
+   if(me->opposite_at_inception.price() <= new_price)
+      detach();            
+   me->num_chases += cross_ticks+1;       
+   if(me->num_chases >= me->max_chase_ticks)
+      detach();   
    report_state();
 }
 
-void LeanOrder::paydown(const md::PriceSize& opposite, uint32_t cross_ticks)
+void LeanOrder::paydown(const md::PriceSize&, const bts::orderman::BookUpdateEvent& book, uint32_t cross_ticks /*= 0*/)
 {
-   double new_price = opposite.price() - tick_size() * cross_ticks;
+   double new_price = me->opposite_at_inception.price() - tick_size() * cross_ticks; 
    double limit = me->opposite_at_inception.price() - tick_size() * me->max_chase_ticks;    
    if(new_price >= limit)
       update_child(Price(new_price), Quantity(this_state().quantity()));
@@ -60,10 +78,50 @@ void LeanOrder::paydown(const md::PriceSize& opposite, uint32_t cross_ticks)
    {
       update_child(Price(limit), Quantity(this_state().quantity()));      
       detach();
-   }
+      return;
+   }            
+   
+   const bts::orderman::BookSide& buybook = book.book.side(api::Side::SIDE_BUY);
+   for(const auto& i : buybook)
+   {
+      me->opposite_at_inception = i;
+      if(me->opposite_at_inception.price() < new_price)
+         break;
+   }               
+   if(me->opposite_at_inception.price() >= new_price)
+      detach();               
+   me->num_chases += cross_ticks+1;       
+   if(me->num_chases >= me->max_chase_ticks)
+      detach();   
    report_state();
 }
 
+void LeanOrder::choose_trade_toward_behavior(const md::PriceSize& opposite)
+{
+   switch (me->trade_toward_behavior)
+   {
+      case LeanOrderDefinition_OnToward_CANCEL:
+      {      
+         cancel_child();
+         transition(ORDER_STATUS_CANCELED);
+         report_state();
+      }
+      break;
+      case LeanOrderDefinition_OnToward_DETACH:
+      {
+         detach();
+      }
+      break;
+      case LeanOrderDefinition_OnToward_REPRICE:
+      {
+         me->opposite_at_inception = opposite;
+      }
+      break;
+      case LeanOrderDefinition_OnToward_NO_CHANGE: 
+      default:
+      break;      
+   }   
+}
 
 void LeanOrder::on_book_update(const bts::orderman::BookUpdateEvent& book)
 {
@@ -91,26 +149,26 @@ void LeanOrder::on_book_update(const bts::orderman::BookUpdateEvent& book)
             {
                const auto& opposite = book.book.side(api::Side::SIDE_SELL)[0];
                if(opposite.price() < me->opposite_at_inception.price())
-                  me->opposite_at_inception = opposite; //your lucky day, try to get luckier
+                  choose_trade_toward_behavior(opposite);
                else if(opposite.price() > me->opposite_at_inception.price())
-                  payup(opposite); //you missed, payup
+                  payup(opposite, book); //you missed, payup/join
                else
                {
                   if(opposite.size() <= thresh.value())
-                     payup(opposite);
+                     payup(opposite, book);
                }
             }
             else
             {
                const auto& opposite = book.book.side(api::Side::SIDE_BUY)[0];
                if(opposite.price() > me->opposite_at_inception.price())
-                  me->opposite_at_inception = opposite; 
+                  choose_trade_toward_behavior(opposite);
                else if(opposite.price() < me->opposite_at_inception.price())
-                  paydown(opposite);
+                  paydown(opposite, book);
                else
                {
                   if(opposite.size() <= thresh.value())
-                     paydown(opposite);
+                     paydown(opposite, book);
                }
             }
          }               

src/lib/orderman/orders/book_algos/LeanOrder.h

@@ -24,8 +24,9 @@ protected:
                     api::orderman::OrderState&, api::orderman::OrderDefinition*) override;
    void on_book_update(const bts::orderman::BookUpdateEvent&) override;   
 private: 
-   void payup(const md::PriceSize& opposite, uint32_t cross_ticks = 0) override;
-   void paydown(const md::PriceSize& opposite, uint32_t cross_ticks = 0) override;
+   void payup(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks = 0) override;
+   void paydown(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks = 0) override;
+   void choose_trade_toward_behavior(const md::PriceSize& opposite);
    
 private:   
    struct PImpl;

src/lib/orderman/orders/book_algos/TickOrder.cpp

@@ -43,7 +43,7 @@ void TickOrder::setup_child(const api::orderman::OrderDefinition& this_def, cons
    me->detach_on_toward = tick_def.detach_on_toward();   
 }
 
-void TickOrder::payup(const md::PriceSize& opposite, uint32_t cross_ticks)
+void TickOrder::payup(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks)
 {
    double new_price = opposite.price() + tick_size() * cross_ticks;
    double limit = me->opposite_at_inception.price() + tick_size() * me->max_chase_ticks;    
@@ -57,7 +57,7 @@ void TickOrder::payup(const md::PriceSize& opposite, uint32_t cross_ticks)
    report_state();
 }
 
-void TickOrder::paydown(const md::PriceSize& opposite, uint32_t cross_ticks)
+void TickOrder::paydown(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks)
 {
    double new_price = opposite.price() - tick_size() * cross_ticks;
    double limit = me->opposite_at_inception.price() - tick_size() * me->max_chase_ticks;    
@@ -127,11 +127,11 @@ void TickOrder::on_book_update(const bts::orderman::BookUpdateEvent& book)
                      me->opposite_at_inception = opposite;                     
                }
                else if(opposite.price() > me->opposite_at_inception.price())
-                  payup(opposite, me->cross_ticks);
+                  payup(opposite, book, me->cross_ticks);
                else
                {
                   if(opposite.size() <= thresh.value())
-                     payup(opposite, me->cross_ticks);
+                     payup(opposite, book, me->cross_ticks);
                }
             }
             else
@@ -145,11 +145,11 @@ void TickOrder::on_book_update(const bts::orderman::BookUpdateEvent& book)
                      me->opposite_at_inception = opposite;
                }
                else if(opposite.price() < me->opposite_at_inception.price())
-                  paydown(opposite, me->cross_ticks);
+                  paydown(opposite, book, me->cross_ticks);
                else
                {
                   if(opposite.size() <= thresh.value())
-                     paydown(opposite, me->cross_ticks);
+                     paydown(opposite, book, me->cross_ticks);
                }
             }
          }               

src/lib/orderman/orders/book_algos/TickOrder.h

@@ -24,8 +24,8 @@ protected:
                     api::orderman::OrderState&, api::orderman::OrderDefinition*) override;
    void on_book_update(const bts::orderman::BookUpdateEvent&) override;   
 private:
-   void payup(const md::PriceSize& opposite, uint32_t cross_ticks) override;
-   void paydown(const md::PriceSize& opposite, uint32_t cross_ticks) override;
+   void payup(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks) override;
+   void paydown(const md::PriceSize& opposite, const bts::orderman::BookUpdateEvent&, uint32_t cross_ticks) override;
    
 private:   
    struct PImpl;

src/lib/orderman/orders/book_algos/TopOrder.cpp

@@ -80,6 +80,33 @@ void TopOrder::on_book_update(const bts::orderman::BookUpdateEvent& book)
       {
          case ORDER_STATUS_INACTIVE: 
          {
+            Quantity thresh = SpreaderUtils::_quantity_threshold(me->top_threshold, Quantity(this_state().quantity()));
+            if(definition().side() == api::SIDE_BUY)
+            {
+               const auto& same = book.book.side(api::Side::SIDE_BUY)[0];
+               if(securities::test_price_equality(this_state().price(), same.price(), tick_size()))
+               {
+                  if(same.size() <= thresh.value())
+                  {
+                     transition(ORDER_STATUS_CANCELED);
+                     report_state();
+                     return;
+                  }
+               }
+            }
+            else
+            {
+               const auto& same = book.book.side(api::Side::SIDE_SELL)[0];
+               if(securities::test_price_equality(this_state().price(), same.price(), tick_size()))
+               {
+                  if(same.size() <= thresh.value())
+                  {
+                     transition(ORDER_STATUS_CANCELED);
+                     report_state();
+                     return;
+                  }
+               }
+            }            
             activate_child(Price(this_state().price()), Quantity(this_state().quantity()));               
             transition(ORDER_STATUS_WORKING);
             report_state();    

src/lib/orderman/perf_analysis/logs/parse_helper.py

@@ -12,6 +12,7 @@ def gen_create_table(d,t):
     print ".import %s trace" % t
     print '''create view tick_to_order as select 
     event_id,
+    order_key,
     (fix_socket_w - md) / 1000 total, 
     (md_src-md)/1000 md_to_md_src, 
     (spreader_on_book-md_src)/1000 md_src_to_spreader, 

src/server/sparkd/http/ApiGateway.cpp

@@ -151,10 +151,12 @@ std::string get_server_info(const api::ServiceContext& ctx) {
 
    if (response.success()) {
       ss << "  \"users\": [" << endl;
+      bool first = true;
       for (const auto& user : response.result()->users()) {
          if (user.is_present()) {
             ss << "    ";
-            if (&user != &response.result()->users()[0]) ss << ",";
+            if (first) { first = false; }
+            else { ss << ","; }
             ss << "{ ";
             ss << "\"name\": \"" << user.name() << "\", ";
             ss << "\"operator_id\": \"" << user.user_data().exchange_user_data().cme_data().operator_id() << "\"";

tools/udp.py

@@ -0,0 +1,29 @@
+#!/usr/bin/python
+
+import sys, os
+import socket, binascii
+
+if len(sys.argv) < 3:
+   print "Usage: udp.py <group> <port> <listen address>"
+   exit(0)
+mcast_grp = sys.argv[1]
+mcast_port = int(sys.argv[2])
+listen_addr = len(sys.argv) == 4 and sys.argv[3] or '0.0.0.0'
+
+sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 32)
+sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
+sock.bind((mcast_grp, mcast_port))
+
+host = listen_addr
+sock.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_IF, socket.inet_aton(host))
+sock.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP, socket.inet_aton(mcast_grp) + socket.inet_aton(host))
+
+while 1:
+   try:
+      data, addr = sock.recvfrom(1024)
+   except socket.error, e:
+      print "error", e
+   hexdata = binascii.hexlify(data)
+   print 'Data = %s' % hexdata