ccxt · caoilainnl · Jun 2, 2025 · Jun 2, 2025 · Jun 10, 2025 · Jun 10, 2025
@@ -17,6 +17,9 @@ public partial class Exchange
     public string userAgent { get; set; }
     public bool verbose { get; set; } = true;
     public bool enableRateLimit { get; set; } = true;
+    public string rateLimiterAlgorithm {get; set; } = "leakyBucket";  // rollingWindow or leakyBucket
+    public int maxLimiterRequests {get; set; } = 1000;
+    public int rollingWindowSize {get; set; } = 60000;
     public long lastRestRequestTimestamp { get; set; } = 0;
     public string url { get; set; } = "";
 
@@ -292,5 +295,8 @@ void initializeProperties(dict userConfig = null)
         this.newUpdates = SafeValue(extendedProperties, "newUpdates") as bool? ?? true;
         this.accounts = SafeValue(extendedProperties, "accounts") as List<object>;
         this.features = SafeValue(extendedProperties, "features", features) as dict;
+        this.rateLimiterAlgorithm = SafeString(extendedProperties, "rateLimiterAlgorithm", "leakyBucket");
+        this.maxLimiterRequests = (int)(SafeInteger(extendedProperties, "maxLimiterRequests") ?? 1000);
+        this.rollingWindowSize = (int)(SafeInteger(extendedProperties, "rollingWindowSize") ?? 60000);
     }
 }
@@ -3,30 +3,49 @@
 namespace ccxt;
 
 using dict = Dictionary<string, object>;
-public class Throttler
+
+public class ThrottlerConfig
 {
+    public double RefillRate { get; set; } = 1.0;
+    public double Delay { get; set; } = 0.001;
+    public double Cost { get; set; } = 1.0;
+    public double Tokens { get; set; } = 0.0;
+    public int MaxLimiterRequests { get; set; } = 2000;
+    public double Capacity { get; set; } = 1.0;
+    public string Algorithm { get; set; } = "leakyBucket";
+    public double RateLimit { get; set; } = 0.0;
+    public double WindowSize { get; set; } = 60000.0;
+    public double MaxWeight { get; set; }
+}
 
-    private dict config = new dict();
+public class Throttler
+{
+    private ThrottlerConfig config = new ThrottlerConfig();
     private Queue<(Task, double)> queue = new Queue<(Task, double)>();
+    private readonly object queueLock = new object();
 
     private bool running = false;
+    private List<(long timestamp, double cost)> timestamps = new List<(long, double)>();
 
-    public Throttler(dict config)
+    public Throttler(Dictionary<string, object> configInput)
     {
-        this.config = new Dictionary<string, object>()
+        // Convert the dictionary input to our typed config
+        if (configInput != null)
         {
-            {"refillRate",1.0},
-            {"delay", 0.001},
-            {"cost", 1.0},
-            {"tokens", 0},
-            {"maxCapacity", 2000},
-            {"capacity", 1.0},
-        };
-        this.config = extend(this.config, config);
-
+            config.RefillRate = configInput.TryGetValue("refillRate", out var refillRate) ? Convert.ToDouble(refillRate) : config.RefillRate;
+            config.Delay = configInput.TryGetValue("delay", out var delay) ? Convert.ToDouble(delay) : config.Delay;
+            config.Cost = configInput.TryGetValue("cost", out var cost) ? Convert.ToDouble(cost) : config.Cost;
+            config.Tokens = configInput.TryGetValue("tokens", out var tokens) ? Convert.ToDouble(tokens) : config.Tokens;
+            config.MaxLimiterRequests = configInput.TryGetValue("maxLimiterRequests", out var maxRequests) ? Convert.ToInt32(maxRequests) : config.MaxLimiterRequests;
+            config.Capacity = configInput.TryGetValue("capacity", out var capacity) ? Convert.ToDouble(capacity) : config.Capacity;
+            config.Algorithm = configInput.TryGetValue("algorithm", out var algorithm) ? Convert.ToString(algorithm) : config.Algorithm;
+            config.RateLimit = configInput.TryGetValue("rateLimit", out var rateLimit) ? Convert.ToDouble(rateLimit) : config.RateLimit;
+            config.WindowSize = configInput.TryGetValue("windowSize", out var windowSize) ? Convert.ToDouble(windowSize) : config.WindowSize;
+            config.MaxWeight = configInput.TryGetValue("maxWeight", out var maxWeight) ? Convert.ToDouble(maxWeight) : config.MaxWeight;
+        }
     }
 
-    private async Task loop()
+    private async Task leakyBucketLoop()
     {
         var lastTimestamp = milliseconds();
         while (this.running)
@@ -40,49 +59,112 @@ private async Task loop()
             var first = this.queue.Peek();
             var task = first.Item1;
             var cost = first.Item2;
-            var tokensAsString = Convert.ToString(this.config["tokens"], CultureInfo.InvariantCulture);
-            var floatTokens = double.Parse(tokensAsString, CultureInfo.InvariantCulture);
+            var floatTokens = this.config.Tokens;
             if (floatTokens >= 0)
             {
-                this.config["tokens"] = floatTokens - cost;
-                await Task.Delay(0);
+                this.config.Tokens = floatTokens - cost;
                 if (task != null)
                 {
                     if (task.Status == TaskStatus.Created)
                     {
                         task.Start();
                     }
                 }
-                this.queue.Dequeue();
-
-                if (this.queue.Count == 0)
+                await Task.Delay(0);
+                lock (queueLock)
                 {
-                    this.running = false;
+                    this.queue.Dequeue();
+                    if (this.queue.Count == 0)
+                    {
+                        this.running = false;
+                    }
                 }
             }
             else
             {
-                await Task.Delay((int)((double)this.config["delay"] * 1000));
+                await Task.Delay((int)(this.config.Delay * 1000));
                 var current = milliseconds();
                 var elapsed = current - lastTimestamp;
                 lastTimestamp = current;
-                var tokens = (double)this.config["tokens"] + ((double)this.config["refillRate"] * elapsed);
-                this.config["tokens"] = Math.Min(tokens, (int)this.config["capacity"]);
+                var tokens = (double)this.config.Tokens + ((double)this.config.RefillRate * elapsed);
+                this.config.Tokens = Math.Min(tokens, (double)this.config.Capacity);
             }
         }
-
     }
 
-    public async Task<Task> throttle(object cost2)
+    private async Task rollingWindowLoop()
     {
+        while (this.running)
+        {
+            lock (queueLock)
+            {
+                if (this.queue.Count == 0)
+                {
+                    this.running = false;
+                    continue;
+                }
+            }
+            (Task, double) first;
+            lock (queueLock)
+            {
+                first = this.queue.Peek();
+            }
+            var task = first.Item1;
+            var cost = first.Item2;
+            var now = milliseconds();
+            var windowSize = this.config.WindowSize;
+            timestamps.RemoveAll(t => now - t.timestamp >= windowSize);
+            var totalCost = timestamps.Sum(t => t.cost);
+            if (totalCost + cost <= this.config.MaxWeight)
+            {
+                timestamps.Add((now, cost));
+                if (task != null && task.Status == TaskStatus.Created)
+                {
+                    task.Start();
+                }
+                await Task.Delay(0);
+                lock (queueLock)
+                {
+                    this.queue.Dequeue();
+                    if (this.queue.Count == 0) this.running = false;
+                }
+            }
+            else
+            {
+                var earliest = timestamps[0].timestamp;
+                var waitTime = (earliest + windowSize) - now;
+                if (waitTime > 0)
+                {
+                    await Task.Delay((int)waitTime);
+                }
+            }
+        }
+    }
 
-        var cost = (cost2 != null) ? Convert.ToDouble(cost2) : Convert.ToDouble(this.config["cost"]);
-        if (this.queue.Count > (int)this.config["maxCapacity"])
+    private async Task loop()
+    {
+        if (this.config.Algorithm == "leakyBucket")
         {
-            throw new Exception("throttle queue is over maxCapacity (" + this.config["maxCapacity"].ToString() + "), see https://github.com/ccxt/ccxt/issues/11645#issuecomment-1195695526");
+            await leakyBucketLoop();
         }
+        else
+        {
+            await rollingWindowLoop();
+        }
+    }
+
+    public async Task<Task> throttle(object cost2)
+    {
+        var cost = (cost2 != null) ? Convert.ToDouble(cost2) : Convert.ToDouble(this.config.Cost);
         var t = new Task(() => { });
-        this.queue.Enqueue((t, cost));
+        lock (queueLock)
+        {
+            if (this.queue.Count > (int)this.config.MaxLimiterRequests)
+            {
+                throw new Exception("throttle queue is over maxLimiterRequests (" + this.config.MaxLimiterRequests.ToString() + "), see https://github.com/ccxt/ccxt/issues/11645#issuecomment-1195695526");
+            }
+            this.queue.Enqueue((t, cost));
+        }
         if (!this.running)
         {
             this.running = true;

@@ -51,6 +51,9 @@ type Exchange struct {
 	Timeout                int64
 	MAX_VALUE              float64
 	RateLimit              float64
+	RateLimiterAlgorithm   string  // rollingWindow or leakyBucket
+	MaxLimiterRequests     int
+	RollingWindowSize      int
 	TokenBucket            map[string]interface{}
 	Throttler              Throttler
 	NewUpdates             bool

@@ -380,6 +380,9 @@ func (this *Exchange) initializeProperties(extendedProperties map[string]interfa
 	}
 	this.EnableRateLimit = SafeValue(extendedProperties, "enableRateLimit", true).(bool)
 	this.RateLimit = SafeFloat(extendedProperties, "rateLimit", -1).(float64)
+	this.RateLimiterAlgorithm = SafeString(extendedProperties, "rateLimiterAlgorithm", "leakyBucket").(string)
+	this.MaxLimiterRequests = int(SafeInteger(extendedProperties, "maxLimiterRequests", 1000).(int64))
+	this.RollingWindowSize = int(SafeInteger(extendedProperties, "rollingWindowSize", 60000).(int64))
 	// this.status = SafeValue(extendedProperties, "status",map[string]interface{}{}).(map[string]interface{})
 	this.PrecisionMode = int(SafeInteger(extendedProperties, "precisionMode", this.PrecisionMode).(int64))
 	this.PaddingMode = int(SafeInteger(extendedProperties, "paddingMode", this.PaddingMode).(int64))