Tilt after a slipped stop: recovery script

[FTtrader]

Here is the institutional execution architecture adapted for cTrader (cAlgo).

Because cTrader operates on C# and .NET, the implementation is significantly cleaner than MQL. We can leverage event-driven architecture (Positions.Closed), LINQ for the statistical modeling, and a Queue<double> for memory-efficient rolling arrays.

A critical nuance in the cTrader API is that once a position hits its Stop Loss, the Position.StopLoss property is often nullified in the closed event. To bypass this, this cBot caches the active Stop Loss of all open positions in a Dictionary, allowing it to accurately compare the expected execution against the actual ClosingPrice found in the History ledger.

cTrader cBot: Quantitative Slippage & State Machine

Code: Select all

using System;
using System.Linq;
using System.Collections.Generic;
using cAlgo.API;
using cAlgo.API.Internals;

namespace cAlgo.Robots
{
    [Robot(TimeZone = TimeZones.UTC, AccessRights = AccessRights.None)]
    public class InstitutionalCircuitBreaker : Robot
    {
        public enum ExecutionState
        {
            Active,     // Full Risk
            Locked,     // Hard Lockout
            Recovery    // Reduced Risk
        }

        // --- System Parameters ---
        [Parameter("Lockout Duration (Min)", Group = "Risk Management", DefaultValue = 15)]
        public int LockoutMinutes { get; set; }

        [Parameter("Z-Score Limit (StdDev)", Group = "Risk Management", DefaultValue = 2.0)]
        public double StdDevLimit { get; set; }

        [Parameter("Min Slip to Track (Pips)", Group = "Risk Management", DefaultValue = 1.0)]
        public double MinSlipPips { get; set; }

        [Parameter("Rolling Array Size", Group = "Risk Management", DefaultValue = 50)]
        public int MaxHistorySize { get; set; }

        // --- State Machine ---
        private ExecutionState _currentState = ExecutionState.Active;
        private DateTime _lockoutEndTime;
        
        // --- Statistical Arrays & Cache ---
        private readonly Queue<double> _slipHistory = new Queue<double>();
        private readonly Dictionary<int, double> _activeStops = new Dictionary<int, double>();

        protected override void OnStart()
        {
            // Subscribe to position events
            Positions.Opened += OnPositionOpened;
            Positions.Modified += OnPositionModified;
            Positions.Closed += OnPositionClosed;

            // Use a 1-second timer to manage state transitions and UI updates
            Timer.Start(1);
            
            Print("Execution Architecture Initialized.");
        }

        protected override void OnTimer()
        {
            // 1. Manage State Unlocking
            if (_currentState == ExecutionState.Locked && Server.Time >= _lockoutEndTime)
            {
                _currentState = ExecutionState.Recovery;
                Print("SYSTEM UNLOCKED: Entering Recovery State (Fractional Risk).");
            }

            // 2. Update UI Dashboard
            UpdateDashboard();
        }

        // --- Stop Loss Caching Engine ---
        private void OnPositionOpened(PositionOpenedEventArgs args) => CacheStopLoss(args.Position);
        private void OnPositionModified(PositionModifiedEventArgs args) => CacheStopLoss(args.Position);

        private void CacheStopLoss(Position position)
        {
            if (position.SymbolName != SymbolName || position.Label != "Institutional_PA") return;

            if (position.StopLoss.HasValue)
                _activeStops[position.Id] = position.StopLoss.Value;
            else
                _activeStops.Remove(position.Id);
        }

        // --- Execution Audit & Variance Math ---
        private void OnPositionClosed(PositionClosedEventArgs args)
        {
            var position = args.Position;
            if (position.SymbolName != SymbolName) return;

            // Clean up the cache
            _activeStops.TryGetValue(position.Id, out double expectedStop);
            _activeStops.Remove(position.Id);

            // Fetch actual execution price from History ledger
            var historicalTrade = History.LastOrDefault(x => x.PositionId == position.Id);
            if (historicalTrade == null) return;

            double closingPrice = historicalTrade.ClosingPrice;

            // Audit only losing trades with a defined Stop Loss
            if (historicalTrade.NetProfit < 0 && expectedStop > 0)
            {
                // Verify the closure was likely a Stop Loss hit (not manual)
                bool wasStoppedOut = (position.TradeType == TradeType.Buy && closingPrice <= expectedStop) ||
                                     (position.TradeType == TradeType.Sell && closingPrice >= expectedStop);

                if (wasStoppedOut)
                {
                    double slipDistance = Math.Abs(expectedStop - closingPrice);
                    double slipPips = slipDistance / Symbol.PipSize;

                    if (slipPips > MinSlipPips)
                    {
                        UpdateStatisticalModel(slipPips);
                    }
                }
            }
            else if (historicalTrade.NetProfit > 0 && _currentState == ExecutionState.Recovery)
            {
                // Successful execution in Recovery resets state to Active
                _currentState = ExecutionState.Active;
                Print("RECOVERY COMPLETE: System Restored to Full Risk.");
            }
        }

        private void UpdateStatisticalModel(double newSlip)
        {
            // Enforce Queue Size
            if (_slipHistory.Count >= MaxHistorySize)
            {
                _slipHistory.Dequeue();
            }
            
            _slipHistory.Enqueue(newSlip);

            if (_slipHistory.Count > 5)
            {
                double mean = _slipHistory.Average();
                double sumOfSquares = _slipHistory.Sum(val => Math.Pow(val - mean, 2));
                double stdDev = Math.Sqrt(sumOfSquares / _slipHistory.Count);

                double zScore = (stdDev > 0) ? (newSlip - mean) / stdDev : 0.0;

                if (zScore > StdDevLimit)
                {
                    _currentState = ExecutionState.Locked;
                    _lockoutEndTime = Server.Time.AddMinutes(LockoutMinutes);
                    
                    string msg = string.Format("CIRCUIT BREAKER: Slipped {0:F1} pips. Z-Score: {1:F2}. System Locked.", newSlip, zScore);
                    Print(msg);
                }
            }
        }

        // --- Price Action Execution Framework ---
        protected override void OnTick()
        {
            if (_currentState == ExecutionState.Locked) return;

            // bool validLongSetup = ... (Insert Raw PA Logic Here)
            // bool validShortSetup = ... (Insert Raw PA Logic Here)
            
            // double riskMultiplier = _currentState == ExecutionState.Recovery ? 0.5 : 1.0;
            
            // if (validLongSetup)
            // {
            //      ExecuteMarketOrder(TradeType.Buy, SymbolName, VolumeInUnits * riskMultiplier, "Institutional_PA", stopLossPips, takeProfitPips);
            // }
        }

        // --- UI Rendering ---
        private void UpdateDashboard()
        {
            string stateTxt = _currentState == ExecutionState.Active ? "ACTIVE (FULL RISK)" :
                              _currentState == ExecutionState.Locked ? "LOCKED (NO EXECUTION)" : 
                              "RECOVERY (HALF RISK)";

            string color = _currentState == ExecutionState.Active ? "LimeGreen" :
                           _currentState == ExecutionState.Locked ? "Red" : 
                           "Orange";

            string timeToUnlock = _currentState == ExecutionState.Locked 
                                  ? Math.Max(0, (_lockoutEndTime - Server.Time).TotalMinutes).ToString("F1") + " min" 
                                  : "N/A";

            string dashboardText = $"<tspan fill=\"Gray\">SYSTEM STATE:</tspan> <tspan fill=\"{color}\" font-weight=\"bold\">{stateTxt}</tspan>\n" +
                                   $"<tspan fill=\"Gray\">Avg Slip (Pips):</tspan> {(_slipHistory.Any() ? _slipHistory.Average().ToString("F2") : "0.00")}\n" +
                                   $"<tspan fill=\"Gray\">Unlock In:</tspan> {timeToUnlock}";

            Chart.DrawStaticText("SysDash", dashboardText, VerticalAlignment.Bottom, HorizontalAlignment.Right, Color.White);
        }
    }
}

[FTtrader]

To transition this from a standard algorithmic script into an enterprise-grade quantitative execution engine, we need to completely restructure the C# architecture.

When you are dealing with microstructure and millisecond execution, standard collections and LINQ queries create unnecessary Garbage Collection (GC) pressure. A professional module uses interface-driven design, zero-allocation data structures (like ring buffers), asynchronous order routing, and strict separation of concerns.

Here is the architectural overview for the forum response, followed by the highly optimized cTrader module.

[FTtrader]

Don't just build a script to stop you from tilting; build an architecture that actively defends your execution quality at the memory level.

cTrader cBot: Enterprise Microstructure Architecture

This version introduces a strict C# architectural pattern. It eliminates LINQ from the statistical modeling, utilizes a zero-allocation Ring Buffer for rolling standard deviation, heavily utilizes interfaces for dependency segregation, and introduces asynchronous order execution.

Code: Select all

using System;
using cAlgo.API;
using cAlgo.API.Internals;

namespace cAlgo.Robots
{
    [Robot(TimeZone = TimeZones.UTC, AccessRights = AccessRights.None)]
    public class InstitutionalExecutionEngine : Robot
    {
        // --- System Parameters ---
        [Parameter("Lockout Duration (Min)", Group = "Microstructure Risk", DefaultValue = 15)]
        public int LockoutMinutes { get; set; }

        [Parameter("Z-Score Limit (StdDev)", Group = "Microstructure Risk", DefaultValue = 2.0)]
        public double StdDevLimit { get; set; }

        [Parameter("Min Slip to Track (Pips)", Group = "Microstructure Risk", DefaultValue = 1.0)]
        public double MinSlipPips { get; set; }

        [Parameter("Max Allowed Spread (Pips)", Group = "Microstructure Risk", DefaultValue = 2.0)]
        public double MaxSpreadPips { get; set; }

        // --- Core Dependencies ---
        private IStateMonitor _stateMonitor;
        private IExecutionManager _executionManager;
        private RingBufferStatistics _slipStats;

        protected override void OnStart()
        {
            // Dependency Injection (Manual wiring for single-file cBot)
            _slipStats = new RingBufferStatistics(50);
            _stateMonitor = new StateMonitor(LockoutMinutes, StdDevLimit, _slipStats);
            
            // Pass the state monitor into the execution manager to enforce lockouts
            _executionManager = new ExecutionManager(this, _stateMonitor, MaxSpreadPips);

            // Event Subscriptions
            Positions.Closed += OnPositionClosed;
            Timer.Start(1); // 1-second UI & State heartbeat

            Print("Institutional Architecture Initialized: Zero-allocation tracking active.");
        }

        protected override void OnTimer()
        {
            _stateMonitor.CheckTimeTransitions(Server.Time);
            RenderDashboard();
        }

        protected override void OnTick()
        {
            if (_stateMonitor.CurrentState == SystemState.Locked) return;

            // --- Price Action Execution Framework ---
            // bool isValidLong = ... (Raw PA Logic)
            
            // if (isValidLong)
            // {
            //     // Fire and forget asynchronous execution to avoid blocking the tick thread
            //     _executionManager.ExecuteAsync(TradeType.Buy, VolumeInUnits, "Institutional_PA", stopLossPips, takeProfitPips);
            // }
        }

        private void OnPositionClosed(PositionClosedEventArgs args)
        {
            if (args.Position.SymbolName != SymbolName) return;

            var trade = History.LastOrDefault(x => x.PositionId == args.Position.Id);
            if (trade == null) return;

            double expectedStop = args.Position.StopLoss ?? 0;
            double closingPrice = trade.ClosingPrice;

            // Audit losing trades that hit the stop loss
            if (trade.NetProfit < 0 && expectedStop > 0)
            {
                bool stoppedOut = (trade.TradeType == TradeType.Buy && closingPrice <= expectedStop) ||
                                  (trade.TradeType == TradeType.Sell && closingPrice >= expectedStop);

                if (stoppedOut)
                {
                    double slipPips = Math.Abs(expectedStop - closingPrice) / Symbol.PipSize;
                    if (slipPips > MinSlipPips)
                    {
                        _stateMonitor.RegisterSlippageEvent(slipPips, Server.Time);
                    }
                }
            }
            else if (trade.NetProfit > 0)
            {
                _stateMonitor.RegisterProfitableExecution();
            }
        }

        private void RenderDashboard()
        {
            string color = _stateMonitor.CurrentState == SystemState.Active ? "LimeGreen" :
                           _stateMonitor.CurrentState == SystemState.Locked ? "Red" : "Orange";

            string text = $"<tspan fill=\"Gray\">ENGINE STATE:</tspan> <tspan fill=\"{color}\" font-weight=\"bold\">{_stateMonitor.CurrentState.ToString().ToUpper()}</tspan>\n" +
                          $"<tspan fill=\"Gray\">Avg Slip (Pips):</tspan> {_slipStats.Mean:F2}\n" +
                          $"<tspan fill=\"Gray\">Real-time Spread:</tspan> {Symbol.Spread / Symbol.PipSize:F1} pips";

            Chart.DrawStaticText("SysDash", text, VerticalAlignment.Bottom, HorizontalAlignment.Right, Color.White);
        }
    }

    // ========================================================================
    // ARCHITECTURAL COMPONENTS & INTERFACES
    // ========================================================================

    public enum SystemState { Active, Locked, Recovery }

    public interface IStateMonitor
    {
        SystemState CurrentState { get; }
        void CheckTimeTransitions(DateTime serverTime);
        void RegisterSlippageEvent(double slippage, DateTime serverTime);
        void RegisterProfitableExecution();
    }

    public interface IExecutionManager
    {
        void ExecuteAsync(TradeType tradeType, double volume, string label, double slPips, double tpPips);
    }

    // --- State Manager ---
    public class StateMonitor : IStateMonitor
    {
        public SystemState CurrentState { get; private set; } = SystemState.Active;
        
        private readonly int _lockoutMinutes;
        private readonly double _zScoreLimit;
        private readonly RingBufferStatistics _stats;
        private DateTime _lockoutEndTime;

        public StateMonitor(int lockoutMinutes, double zScoreLimit, RingBufferStatistics stats)
        {
            _lockoutMinutes = lockoutMinutes;
            _zScoreLimit = zScoreLimit;
            _stats = stats;
        }

        public void CheckTimeTransitions(DateTime serverTime)
        {
            if (CurrentState == SystemState.Locked && serverTime >= _lockoutEndTime)
            {
                CurrentState = SystemState.Recovery;
            }
        }

        public void RegisterSlippageEvent(double slippage, DateTime serverTime)
        {
            _stats.Add(slippage);

            if (_stats.Count > 5)
            {
                double zScore = _stats.GetZScore(slippage);
                if (zScore > _zScoreLimit)
                {
                    CurrentState = SystemState.Locked;
                    _lockoutEndTime = serverTime.AddMinutes(_lockoutMinutes);
                }
            }
        }

        public void RegisterProfitableExecution()
        {
            if (CurrentState == SystemState.Recovery)
            {
                CurrentState = SystemState.Active; // Restored to full risk
            }
        }
    }

    // --- Asynchronous Execution Engine ---
    public class ExecutionManager : IExecutionManager
    {
        private readonly Robot _robot;
        private readonly IStateMonitor _state;
        private readonly double _maxSpreadPips;

        public ExecutionManager(Robot robot, IStateMonitor state, double maxSpreadPips)
        {
            _robot = robot;
            _state = state;
            _maxSpreadPips = maxSpreadPips;
        }

        public void ExecuteAsync(TradeType tradeType, double volume, string label, double slPips, double tpPips)
        {
            if (_state.CurrentState == SystemState.Locked) return;

            // Microstructure check: Block execution if liquidity is evaporated
            double currentSpread = _robot.Symbol.Spread / _robot.Symbol.PipSize;
            if (currentSpread > _maxSpreadPips)
            {
                _robot.Print($"EXECUTION BLOCKED: Spread ({currentSpread:F1}) exceeds maximum allowed ({_maxSpreadPips:F1}).");
                return;
            }

            double riskVolume = _state.CurrentState == SystemState.Recovery ? volume * 0.5 : volume;

            // Asynchronous routing prevents main thread blocking during PA signal bursts
            _robot.ExecuteMarketOrderAsync(tradeType, _robot.SymbolName, riskVolume, label, slPips, tpPips, response =>
            {
                if (!response.IsSuccessful)
                {
                    _robot.Print($"Order Failed: {response.Error}");
                }
            });
        }
    }

    // --- Zero-Allocation Ring Buffer for Mathematics (O(1) Memory) ---
    public class RingBufferStatistics
    {
        private readonly double[] _buffer;
        private int _index = 0;
        public int Count { get; private set; } = 0;
        public double Mean { get; private set; } = 0;

        public RingBufferStatistics(int size)
        {
            _buffer = new double[size];
        }

        public void Add(double value)
        {
            _buffer[_index] = value;
            _index = (_index + 1) % _buffer.Length;
            if (Count < _buffer.Length) Count++;

            CalculateMean();
        }

        private void CalculateMean()
        {
            double sum = 0;
            for (int i = 0; i < Count; i++) sum += _buffer[i];
            Mean = Count > 0 ? sum / Count : 0;
        }

        public double GetZScore(double value)
        {
            if (Count < 2) return 0;
            
            double sumOfSquares = 0;
            for (int i = 0; i < Count; i++)
            {
                sumOfSquares += Math.Pow(_buffer[i] - Mean, 2);
            }
            
            double stdDev = Math.Sqrt(sumOfSquares / Count);
            return stdDev > 0 ? (value - Mean) / stdDev : 0;
        }
    }
}

[FTtrader]

Logging execution data to an external relational database like MS SQL Server is the logical next step for building a proprietary execution analysis dataset.

However, introducing network I/O operations directly into a high-frequency microstructure bot is dangerous. A synchronous database insert takes milliseconds—an eternity during a liquidity sweep. To maintain the zero-latency architecture of the execution thread, the database logging must be completely decoupled using a Producer-Consumer pattern.

Here is how to architect an asynchronous, non-blocking SQL logger for cTrader.

[FTtrader]

1. Database Schema (MS SQL Server)

First, provision the target table. Using DATETIME2 ensures you capture the exact millisecond of the execution, which is critical for spread and routing analysis.

Code: Select all

CREATE TABLE SlippageAudit (
    EventId INT IDENTITY(1,1) PRIMARY KEY,
    ExecutionTime DATETIME2(3) NOT NULL,
    Symbol NVARCHAR(15) NOT NULL,
    ExpectedStop DECIMAL(18,5) NOT NULL,
    ActualFill DECIMAL(18,5) NOT NULL,
    SlipPips DECIMAL(9,2) NOT NULL,
    SystemState NVARCHAR(20) NOT NULL,
    SpreadAtExecution DECIMAL(9,2) NOT NULL
);

[FTtrader]

2. cTrader Environment Setup

To allow cTrader to communicate with an external database:

Change the bot's access rights in the class attribute to AccessRights = AccessRights.FullAccess. Network requests are blocked by default.

Open the cBot in Visual Studio (or use the cTrader built-in NuGet package manager) and install Microsoft.Data.SqlClient.

3. The Non-Blocking SQL Logger Architecture

This implementation uses System.Threading.Channels, which is a highly optimized, allocation-free queue for Producer-Consumer workflows in .NET. The tick thread drops the execution record into the channel and immediately returns to the market. A dedicated background task drains the channel and executes the SQL INSERT statements.

Code: Select all

using System;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Microsoft.Data.SqlClient;
using cAlgo.API;

namespace cAlgo.Robots
{
    // --- Data Transfer Object ---
    public readonly struct ExecutionRecord
    {
        public DateTime ExecutionTime { get; }
        public string Symbol { get; }
        public double ExpectedStop { get; }
        public double ActualFill { get; }
        public double SlipPips { get; }
        public string SystemState { get; }
        public double SpreadAtExecution { get; }

        public ExecutionRecord(DateTime time, string symbol, double expected, double actual, double slip, string state, double spread)
        {
            ExecutionTime = time;
            Symbol = symbol;
            ExpectedStop = expected;
            ActualFill = actual;
            SlipPips = slip;
            SystemState = state;
            SpreadAtExecution = spread;
        }
    }

    // --- Interface for Dependency Injection ---
    public interface IDatabaseLogger
    {
        void EnqueueRecord(ExecutionRecord record);
        void Stop();
    }

    // --- Asynchronous SQL Logger ---
    public class SqlServerLogger : IDatabaseLogger
    {
        private readonly string _connectionString;
        private readonly Channel<ExecutionRecord> _channel;
        private readonly CancellationTokenSource _cts;
        private readonly Task _workerTask;

        public SqlServerLogger(string connectionString)
        {
            _connectionString = connectionString;
            
            // Create an unbounded channel (or bounded if you want to strictly limit memory)
            _channel = Channel.CreateUnbounded<ExecutionRecord>(new UnboundedChannelOptions
            {
                SingleReader = true,
                SingleWriter = false
            });

            _cts = new CancellationTokenSource();
            _workerTask = Task.Run(() => ProcessQueueAsync(_cts.Token));
        }

        // Called by the cBot tick thread (O(1) complexity, returns instantly)
        public void EnqueueRecord(ExecutionRecord record)
        {
            _channel.Writer.TryWrite(record);
        }

        // Dedicated background I/O thread
        private async Task ProcessQueueAsync(CancellationToken token)
        {
            try
            {
                // Await data without burning CPU cycles
                await foreach (var record in _channel.Reader.ReadAllAsync(token))
                {
                    await InsertToDatabaseAsync(record);
                }
            }
            catch (OperationCanceledException)
            {
                // Graceful shutdown
            }
        }

        private async Task InsertToDatabaseAsync(ExecutionRecord record)
        {
            const string query = @"
                INSERT INTO SlippageAudit 
                (ExecutionTime, Symbol, ExpectedStop, ActualFill, SlipPips, SystemState, SpreadAtExecution) 
                VALUES (@Time, @Symbol, @Expected, @Actual, @Slip, @State, @Spread)";

            try
            {
                using (var conn = new SqlConnection(_connectionString))
                using (var cmd = new SqlCommand(query, conn))
                {
                    cmd.Parameters.AddWithValue("@Time", record.ExecutionTime);
                    cmd.Parameters.AddWithValue("@Symbol", record.Symbol);
                    cmd.Parameters.AddWithValue("@Expected", record.ExpectedStop);
                    cmd.Parameters.AddWithValue("@Actual", record.ActualFill);
                    cmd.Parameters.AddWithValue("@Slip", record.SlipPips);
                    cmd.Parameters.AddWithValue("@State", record.SystemState);
                    cmd.Parameters.AddWithValue("@Spread", record.SpreadAtExecution);

                    await conn.OpenAsync();
                    await cmd.ExecuteNonQueryAsync();
                }
            }
            catch (Exception ex)
            {
                // In production, log this failure to a local flat file so data isn't lost
                Console.WriteLine($"SQL Insert Failed: {ex.Message}");
            }
        }

        public void Stop()
        {
            _channel.Writer.Complete();
            _cts.Cancel();
            _workerTask.Wait(TimeSpan.FromSeconds(3)); // Wait for final writes to flush
        }
    }
}

[FTtrader]

4. Wiring the Logger into the cBot Architecture

To integrate this safely, update the main cBot class. You must instantiate the logger on startup, inject it into the event cycle, and importantly, tear it down gracefully when the bot stops to ensure no pending SQL commands are killed mid-flight.

C#

Code: Select all

[Robot(TimeZone = TimeZones.UTC, AccessRights = AccessRights.FullAccess)]
    public class InstitutionalExecutionEngine : Robot
    {
        [Parameter("SQL Connection String", Group = "Database", DefaultValue = "Server=localhost;Database=AlgoMetrics;User Id=sa;Password=your_password;TrustServerCertificate=True;")]
        public string DbConnectionString { get; set; }

        private IDatabaseLogger _sqlLogger;
        // ... existing dependencies ...

        protected override void OnStart()
        {
            // Initialize the asynchronous logger
            _sqlLogger = new SqlServerLogger(DbConnectionString);
            
            // ... existing startup logic ...
        }

        private void OnPositionClosed(PositionClosedEventArgs args)
        {
            // ... existing slippage calculation ...

            if (stoppedOut)
            {
                double slipPips = Math.Abs(expectedStop - closingPrice) / Symbol.PipSize;
                
                // Fire and forget to the logging channel
                var record = new ExecutionRecord(
                    time: Server.Time,
                    symbol: SymbolName,
                    expected: expectedStop,
                    actual: closingPrice,
                    slip: slipPips,
                    state: _stateMonitor.CurrentState.ToString(),
                    spread: Symbol.Spread / Symbol.PipSize
                );
                
                _sqlLogger.EnqueueRecord(record);

                if (slipPips > MinSlipPips)
                {
                    _stateMonitor.RegisterSlippageEvent(slipPips, Server.Time);
                }
            }
        }

        protected override void OnStop()
        {
            // Crucial: Flushes the queue and gracefully closes DB connections
            _sqlLogger?.Stop();
        }
    }

[FTtrader]

Building a resilient data pipeline requires a Circuit Breaker and a Dead Letter/Fallback Queue. If you are logging execution data in a live market environment, network jitter or database maintenance cannot be allowed to halt your tracking or block the main thread.

To achieve this, we will upgrade the SqlServerLogger to include a local file-based fallback mechanism.

Architectural Approach

Circuit Breaker State: The logger maintains a boolean _isDbOnline flag. If a SQL insert fails, the circuit "trips" and routes all immediate subsequent records directly to a local CSV file, bypassing the network timeout delay.

Atomic File Renaming: To prevent file locking conflicts between the thread writing new offline records and the thread trying to recover them, the recovery task renames the OfflineQueue.csv to a .processing file before reading it.

Background Recovery Task: A secondary background loop periodically checks the database health. If the DB is back online, it reads the .processing file, streams the records into SQL, and deletes the file once successful.

[FTtrader]

Here is the enterprise-grade implementation:

1. The Resilient SQL Logger

This replaces the previous SqlServerLogger class. It now requires System.IO and a file path to store the fallback data.

Code: Select all

using System;
using System.IO;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using Microsoft.Data.SqlClient;
using System.Globalization;

namespace cAlgo.Robots
{
    public class ResilientSqlLogger : IDatabaseLogger
    {
        private readonly string _connectionString;
        private readonly string _fallbackFilePath;
        private readonly string _processingFilePath;
        
        private readonly Channel<ExecutionRecord> _channel;
        private readonly CancellationTokenSource _cts;
        private readonly Task _writerTask;
        private readonly Task _recoveryTask;
        
        private readonly SemaphoreSlim _fileLock = new SemaphoreSlim(1, 1);
        private volatile bool _isDbOnline = true;

        public ResilientSqlLogger(string connectionString, string storageDirectory)
        {
            _connectionString = connectionString;
            _fallbackFilePath = Path.Combine(storageDirectory, "SlippageOfflineQueue.csv");
            _processingFilePath = Path.Combine(storageDirectory, "SlippageOfflineQueue.processing");

            _channel = Channel.CreateUnbounded<ExecutionRecord>(new UnboundedChannelOptions
            {
                SingleReader = true,
                SingleWriter = false
            });

            _cts = new CancellationTokenSource();
            
            // Task 1: Drains the live execution channel
            _writerTask = Task.Run(() => ProcessLiveQueueAsync(_cts.Token));
            
            // Task 2: Periodically checks for and recovers offline records
            _recoveryTask = Task.Run(() => OfflineRecoveryLoopAsync(_cts.Token));
        }

        public void EnqueueRecord(ExecutionRecord record)
        {
            _channel.Writer.TryWrite(record);
        }

        // --- LIVE QUEUE PROCESSING ---
        private async Task ProcessLiveQueueAsync(CancellationToken token)
        {
            try
            {
                await foreach (var record in _channel.Reader.ReadAllAsync(token))
                {
                    if (_isDbOnline)
                    {
                        bool success = await TryInsertToDatabaseAsync(record);
                        if (!success)
                        {
                            _isDbOnline = false; // Trip the circuit breaker
                            await SaveToFallbackAsync(record);
                        }
                    }
                    else
                    {
                        // DB is known offline, route straight to disk
                        await SaveToFallbackAsync(record);
                    }
                }
            }
            catch (OperationCanceledException) { /* Graceful shutdown */ }
        }

        // --- SQL EXECUTION ---
        private async Task<bool> TryInsertToDatabaseAsync(ExecutionRecord record)
        {
            const string query = @"
                INSERT INTO SlippageAudit 
                (ExecutionTime, Symbol, ExpectedStop, ActualFill, SlipPips, SystemState, SpreadAtExecution) 
                VALUES (@Time, @Symbol, @Expected, @Actual, @Slip, @State, @Spread)";

            try
            {
                using (var conn = new SqlConnection(_connectionString))
                using (var cmd = new SqlCommand(query, conn))
                {
                    cmd.Parameters.AddWithValue("@Time", record.ExecutionTime);
                    cmd.Parameters.AddWithValue("@Symbol", record.Symbol);
                    cmd.Parameters.AddWithValue("@Expected", record.ExpectedStop);
                    cmd.Parameters.AddWithValue("@Actual", record.ActualFill);
                    cmd.Parameters.AddWithValue("@Slip", record.SlipPips);
                    cmd.Parameters.AddWithValue("@State", record.SystemState);
                    cmd.Parameters.AddWithValue("@Spread", record.SpreadAtExecution);

                    await conn.OpenAsync();
                    await cmd.ExecuteNonQueryAsync();
                    return true;
                }
            }
            catch (SqlException)
            {
                return false;
            }
        }

        // --- FALLBACK DISK I/O ---
        private async Task SaveToFallbackAsync(ExecutionRecord record)
        {
            await _fileLock.WaitAsync();
            try
            {
                bool writeHeader = !File.Exists(_fallbackFilePath);
                using (var writer = new StreamWriter(_fallbackFilePath, append: true))
                {
                    if (writeHeader) 
                        await writer.WriteLineAsync("Time,Symbol,Expected,Actual,Slip,State,Spread");
                    
                    await writer.WriteLineAsync($"{record.ExecutionTime:O},{record.Symbol},{record.ExpectedStop},{record.ActualFill},{record.SlipPips},{record.SystemState},{record.SpreadAtExecution}");
                }
            }
            finally
            {
                _fileLock.Release();
            }
        }

        // --- RECOVERY LOOP ---
        private async Task OfflineRecoveryLoopAsync(CancellationToken token)
        {
            while (!token.IsCancellationRequested)
            {
                try
                {
                    await Task.Delay(TimeSpan.FromMinutes(1), token); // Check every 60 seconds

                    await _fileLock.WaitAsync(token);
                    try
                    {
                        if (File.Exists(_fallbackFilePath))
                        {
                            // Atomically rename to avoid write locks from the live thread
                            if (File.Exists(_processingFilePath)) File.Delete(_processingFilePath);
                            File.Move(_fallbackFilePath, _processingFilePath);
                        }
                    }
                    finally
                    {
                        _fileLock.Release();
                    }

                    if (File.Exists(_processingFilePath))
                    {
                        bool allRecovered = await ProcessOfflineFileAsync(_processingFilePath);
                        
                        if (allRecovered)
                        {
                            File.Delete(_processingFilePath);
                            _isDbOnline = true; // DB is healthy, reset circuit breaker
                        }
                        else
                        {
                            // If it fails mid-file, rename it back to fallback so we don't lose data
                            await _fileLock.WaitAsync(token);
                            try
                            {
                                // Append remaining processing data back to the main fallback file
                                var remainingData = File.ReadAllText(_processingFilePath);
                                File.AppendAllText(_fallbackFilePath, remainingData);
                                File.Delete(_processingFilePath);
                            }
                            finally
                            {
                                _fileLock.Release();
                            }
                        }
                    }
                    else
                    {
                        // No file exists, assume network is fine
                        _isDbOnline = true; 
                    }
                }
                catch (OperationCanceledException) { break; }
                catch (Exception) { /* Swallow parsing errors to keep loop alive */ }
            }
        }

        private async Task<bool> ProcessOfflineFileAsync(string filePath)
        {
            var lines = File.ReadAllLines(filePath);
            
            // Start from 1 to skip the CSV header
            for (int i = 1; i < lines.Length; i++) 
            {
                if (string.IsNullOrWhiteSpace(lines[i])) continue;
                
                var cols = lines[i].Split(',');
                if (cols.Length < 7) continue;

                var record = new ExecutionRecord(
                    time: DateTime.Parse(cols[0], null, DateTimeStyles.RoundtripKind),
                    symbol: cols[1],
                    expected: double.Parse(cols[2], CultureInfo.InvariantCulture),
                    actual: double.Parse(cols[3], CultureInfo.InvariantCulture),
                    slip: double.Parse(cols[4], CultureInfo.InvariantCulture),
                    state: cols[5],
                    spread: double.Parse(cols[6], CultureInfo.InvariantCulture)
                );

                bool success = await TryInsertToDatabaseAsync(record);
                if (!success)
                {
                    // Rewrite the file from the current failure point onward
                    File.WriteAllLines(filePath, lines[i..^0]);
                    return false;
                }
            }
            return true;
        }

        public void Stop()
        {
            _channel.Writer.Complete();
            _cts.Cancel();
            Task.WaitAll(new[] { _writerTask, _recoveryTask }, TimeSpan.FromSeconds(5));
        }
    }
}

[FTtrader]

2. Integration into the cBot

To pass a safe directory for the CSV file, use Environment.GetFolderPath combined with your application name, or leverage cTrader's native storage paths if available.

Update the OnStart method of your main cBot:

Code: Select all

protected override void OnStart()
{
    // Define a safe local path for the fallback CSV
    string appData = Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData);
    string storageDir = Path.Combine(appData, "cTrader_InstitutionalLogger");
    
    if (!Directory.Exists(storageDir))
    {
        Directory.CreateDirectory(storageDir);
    }

    // Initialize the resilient logger
    _sqlLogger = new ResilientSqlLogger(DbConnectionString, storageDir);
    
    // ... initialize the rest of the architecture ...
}

This architecture guarantees that the tick thread remains completely unblocked, memory allocations remain virtually zero during standard operation, and your execution data is never lost due to external infrastructure failures.