<commit_before>﻿// -----------------------------------------------------------------------
// <copyright file="IB.cs" company="">
// Copyright 2013 <NAME>
// </copyright>
// -----------------------------------------------------------------------

// The IB system is a bit weird, with limitations both on the amount of data
// that can be requested in a single request, and on the frequency of requests.
// The system keeps track of requests, and if they fail due to pacing constraints
// resends them using the _requestRepeatTimer Elapsed event.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Timers;
using System.Windows;
using Krs.Ats.IBNet;
using NLog;
using QDMS;
using LogLevel = NLog.LogLevel;

namespace QDMSServer.DataSources
{
    internal class IB : IHistoricalDataSource, IRealTimeDataSource
    {
        private readonly IBClient _client;
        private readonly Dictionary<int, RealTimeDataRequest> _realTimeDataRequests;
        private readonly ConcurrentDictionary<int, HistoricalDataRequest> _historicalDataRequests;

        /// <summary>
        /// Sub-requests are created when we need to send multiple requests to the 
        /// IB client to fulfill a single data request. This one holds the ID mappings between them.
        /// Key: sub-request ID, Value: the ID of the original request that generated it.
        /// </summary>
        private readonly Dictionary<int, int> _subRequestIDMap;

        /// <summary>
        /// This holds the number of outstanding sub requests.
        /// Key: original request ID, Value: the number of subrequests sent out but not returned.
        /// </summary>
        private readonly Dictionary<int, int> _subRequestCount;

        private readonly Queue<int> _realTimeRequestQueue;
        private readonly Queue<int> _historicalRequestQueue;

        private readonly ConcurrentDictionary<int, List<OHLCBar>> _arrivedHistoricalData;

        private readonly Timer _requestRepeatTimer;
        private int _requestCounter;
        private readonly Logger _logger = LogManager.GetCurrentClassLogger();

        private readonly int _clientID;
        private readonly object _queueLock = new object();
        private readonly object _subReqMapLock = new object();

        public string Name { get; private set; }

        public bool Connected { get { return _client.Connected; } }

        public IB(int clientID = -1)
        {
            Name = "Interactive Brokers";

            if (clientID < 0)
                clientID = Properties.Settings.Default.ibClientID;
            _clientID = clientID;

            _realTimeDataRequests = new Dictionary<int, RealTimeDataRequest>();
            _realTimeRequestQueue = new Queue<int>();

            _historicalDataRequests = new ConcurrentDictionary<int, HistoricalDataRequest>();
            _historicalRequestQueue = new Queue<int>();

            _arrivedHistoricalData = new ConcurrentDictionary<int, List<OHLCBar>>();

            _subRequestIDMap = new Dictionary<int, int>();
            _subRequestCount = new Dictionary<int, int>();

            _requestRepeatTimer = new Timer(20000); //we wait 20 seconds to repeat failed requests
            _requestRepeatTimer.Elapsed += ReSendRequests;

            _requestCounter = 1;

            _client = new IBClient();
            _client.Error += _client_Error;
            _client.ConnectionClosed += _client_ConnectionClosed;
            _client.RealTimeBar += _client_RealTimeBar;
            _client.HistoricalData += _client_HistoricalData;
        }

        /// <summary>
        /// This event is raised when historical data arrives from TWS
        /// </summary>
        private void _client_HistoricalData(object sender, Krs.Ats.IBNet.HistoricalDataEventArgs e)
        {
            //convert the bar and add it to the Dictionary of arrived data
            var bar = TWSUtils.HistoricalDataEventArgsToOHLCBar(e);
            int id;

            lock (_subReqMapLock)
            {
                //if the data is arriving for a sub-request, we must get the id of the original request first
                //otherwise it's just the same id
                id = _subRequestIDMap.ContainsKey(e.RequestId) 
                    ? _subRequestIDMap[e.RequestId] 
                    : e.RequestId;
            }

            var request = _historicalDataRequests[id];

            //One day or lower frequency means we don't get time data.
            //Instead we provide our own by using that day's session end...
            //perhaps this should be moved to the HistoricalDataBroker instead?
            var exchangeTZ = TimeZoneInfo.FindSystemTimeZoneById(request.Instrument.Exchange.Timezone);
            if (request.Frequency >= QDMS.BarSize.OneDay)
            {
                bar.DT = TimeZoneInfo.ConvertTime(bar.DT, TimeZoneInfo.Local, exchangeTZ);
                var dotw = bar.DT.DayOfWeek.ToInt();
                var daysSessionEnd = request.Instrument.Sessions.FirstOrDefault(x => (int)x.ClosingDay == dotw && x.IsSessionEnd);
                if (daysSessionEnd != null)
                {
                    bar.DT = new DateTime(bar.DT.Year, bar.DT.Month, bar.DT.Day) + daysSessionEnd.ClosingTime;
                }
                else
                {
                    bar.DT = new DateTime(bar.DT.Year, bar.DT.Month, bar.DT.Day, 23, 59, 59);
                }
            }
            else
            {
                bar.DT = TimeZoneInfo.ConvertTime(bar.DT, TimeZoneInfo.Local, exchangeTZ);
            }

            //stocks need to have their volumes multiplied by 100, I think all other instrument types do not
            if (request.Instrument.Type == InstrumentType.Stock)
                bar.Volume *= 100;



            _arrivedHistoricalData[id].Add(bar);

            if (e.RecordNumber >= e.RecordTotal - 1) //this was the last item to receive for this request, send it to the broker
            {
                bool requestComplete = true;

                lock (_subReqMapLock)
                {
                    if (_subRequestIDMap.ContainsKey(e.RequestId))
                    {
                        _subRequestIDMap.Remove(e.RequestId);
                        _subRequestCount[id]--;
                        if (_subRequestCount[id] > 0)
                        {
                            //What happens here: this is a subrequest.
                            //We check how many sub-requests in this group have been delivered.
                            //if this is the last one, we want to call HistoricalDataRequestComplete()
                            //otherwise there's more data to come, so we have to wait for it
                            requestComplete = false;
                        }
                        else
                        {
                            //if it is complete, we'll need to sort the data because subrequests may not come in in order
                            _arrivedHistoricalData[id] = _arrivedHistoricalData[id].OrderBy(x => x.DT).ToList();
                        }
                    }
                }

                if(requestComplete)
                    HistoricalDataRequestComplete(id);
            }
        }

        /// <summary>
        /// This method is called when a historical data request has delivered all its bars
        /// </summary>
        /// <param name="requestID"></param>
        private void HistoricalDataRequestComplete(int requestID)
        {
            //IB doesn't actually allow us to provide a deterministic starting point for our historical data query
            //so sometimes we get data from earlier than we want
            //here we throw it away
            var cutoffDate = _historicalDataRequests[requestID].StartingDate.Date;

            List<OHLCBar> bars;
            var removed = _arrivedHistoricalData.TryRemove(requestID, out bars);
            if (!removed) return;

            //due to the nature of sub-requests, the list may contain the same points multiple times
            //so we grab unique values only
            bars = bars.Distinct((x, y) => x.DT == y.DT).ToList();

            //if the data is daily or lower freq, set adjusted ohlc values for convenience
            if (_historicalDataRequests[requestID].Frequency >= QDMS.BarSize.OneDay)
            {
                foreach (OHLCBar b in bars)
                {
                    b.AdjOpen = b.Open;
                    b.AdjHigh = b.High;
                    b.AdjLow = b.Low;
                    b.AdjClose = b.Close;
                }
            }

            RaiseEvent(HistoricalDataArrived, this, new QDMS.HistoricalDataEventArgs(
                _historicalDataRequests[requestID],
                bars.Where(x => x.DT.Date >= cutoffDate).ToList()));
        }

        //This event is raised when real time data arrives
        //We convert them and pass them on downstream
        private void _client_RealTimeBar(object sender, RealTimeBarEventArgs e)
        {
            RealTimeDataEventArgs args = TWSUtils.RealTimeDataEventArgsConverter(e);
            args.Symbol = _realTimeDataRequests[e.RequestId].Instrument.Symbol;
            RaiseEvent(DataReceived, this, args);
        }

        //This event is raised when the connection to TWS client closed.
        private void _client_ConnectionClosed(object sender, ConnectionClosedEventArgs e)<commit_msg>Only set adjusted values for stocks.<commit_after>﻿// -----------------------------------------------------------------------
// <copyright file="IB.cs" company="">
// Copyright 2013 <NAME>
// </copyright>
// -----------------------------------------------------------------------

// The IB system is a bit weird, with limitations both on the amount of data
// that can be requested in a single request, and on the frequency of requests.
// The system keeps track of requests, and if they fail due to pacing constraints
// resends them using the _requestRepeatTimer Elapsed event.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Timers;
using System.Windows;
using Krs.Ats.IBNet;
using NLog;
using QDMS;
using LogLevel = NLog.LogLevel;

namespace QDMSServer.DataSources
{
    internal class IB : IHistoricalDataSource, IRealTimeDataSource
    {
        private readonly IBClient _client;
        private readonly Dictionary<int, RealTimeDataRequest> _realTimeDataRequests;
        private readonly ConcurrentDictionary<int, HistoricalDataRequest> _historicalDataRequests;

        /// <summary>
        /// Sub-requests are created when we need to send multiple requests to the 
        /// IB client to fulfill a single data request. This one holds the ID mappings between them.
        /// Key: sub-request ID, Value: the ID of the original request that generated it.
        /// </summary>
        private readonly Dictionary<int, int> _subRequestIDMap;

        /// <summary>
        /// This holds the number of outstanding sub requests.
        /// Key: original request ID, Value: the number of subrequests sent out but not returned.
        /// </summary>
        private readonly Dictionary<int, int> _subRequestCount;

        private readonly Queue<int> _realTimeRequestQueue;
        private readonly Queue<int> _historicalRequestQueue;

        private readonly ConcurrentDictionary<int, List<OHLCBar>> _arrivedHistoricalData;

        private readonly Timer _requestRepeatTimer;
        private int _requestCounter;
        private readonly Logger _logger = LogManager.GetCurrentClassLogger();

        private readonly int _clientID;
        private readonly object _queueLock = new object();
        private readonly object _subReqMapLock = new object();

        public string Name { get; private set; }

        public bool Connected { get { return _client.Connected; } }

        public IB(int clientID = -1)
        {
            Name = "<NAME>";

            if (clientID < 0)
                clientID = Properties.Settings.Default.ibClientID;
            _clientID = clientID;

            _realTimeDataRequests = new Dictionary<int, RealTimeDataRequest>();
            _realTimeRequestQueue = new Queue<int>();

            _historicalDataRequests = new ConcurrentDictionary<int, HistoricalDataRequest>();
            _historicalRequestQueue = new Queue<int>();

            _arrivedHistoricalData = new ConcurrentDictionary<int, List<OHLCBar>>();

            _subRequestIDMap = new Dictionary<int, int>();
            _subRequestCount = new Dictionary<int, int>();

            _requestRepeatTimer = new Timer(20000); //we wait 20 seconds to repeat failed requests
            _requestRepeatTimer.Elapsed += ReSendRequests;

            _requestCounter = 1;

            _client = new IBClient();
            _client.Error += _client_Error;
            _client.ConnectionClosed += _client_ConnectionClosed;
            _client.RealTimeBar += _client_RealTimeBar;
            _client.HistoricalData += _client_HistoricalData;
        }

        /// <summary>
        /// This event is raised when historical data arrives from TWS
        /// </summary>
        private void _client_HistoricalData(object sender, Krs.Ats.IBNet.HistoricalDataEventArgs e)
        {
            //convert the bar and add it to the Dictionary of arrived data
            var bar = TWSUtils.HistoricalDataEventArgsToOHLCBar(e);
            int id;

            lock (_subReqMapLock)
            {
                //if the data is arriving for a sub-request, we must get the id of the original request first
                //otherwise it's just the same id
                id = _subRequestIDMap.ContainsKey(e.RequestId) 
                    ? _subRequestIDMap[e.RequestId] 
                    : e.RequestId;
            }

            var request = _historicalDataRequests[id];

            //One day or lower frequency means we don't get time data.
            //Instead we provide our own by using that day's session end...
            //perhaps this should be moved to the HistoricalDataBroker instead?
            var exchangeTZ = TimeZoneInfo.FindSystemTimeZoneById(request.Instrument.Exchange.Timezone);
            if (request.Frequency >= QDMS.BarSize.OneDay)
            {
                bar.DT = TimeZoneInfo.ConvertTime(bar.DT, TimeZoneInfo.Local, exchangeTZ);
                var dotw = bar.DT.DayOfWeek.ToInt();
                var daysSessionEnd = request.Instrument.Sessions.FirstOrDefault(x => (int)x.ClosingDay == dotw && x.IsSessionEnd);
                if (daysSessionEnd != null)
                {
                    bar.DT = new DateTime(bar.DT.Year, bar.DT.Month, bar.DT.Day) + daysSessionEnd.ClosingTime;
                }
                else
                {
                    bar.DT = new DateTime(bar.DT.Year, bar.DT.Month, bar.DT.Day, 23, 59, 59);
                }
            }
            else
            {
                bar.DT = TimeZoneInfo.ConvertTime(bar.DT, TimeZoneInfo.Local, exchangeTZ);
            }

            //stocks need to have their volumes multiplied by 100, I think all other instrument types do not
            if (request.Instrument.Type == InstrumentType.Stock)
                bar.Volume *= 100;



            _arrivedHistoricalData[id].Add(bar);

            if (e.RecordNumber >= e.RecordTotal - 1) //this was the last item to receive for this request, send it to the broker
            {
                bool requestComplete = true;

                lock (_subReqMapLock)
                {
                    if (_subRequestIDMap.ContainsKey(e.RequestId))
                    {
                        _subRequestIDMap.Remove(e.RequestId);
                        _subRequestCount[id]--;
                        if (_subRequestCount[id] > 0)
                        {
                            //What happens here: this is a subrequest.
                            //We check how many sub-requests in this group have been delivered.
                            //if this is the last one, we want to call HistoricalDataRequestComplete()
                            //otherwise there's more data to come, so we have to wait for it
                            requestComplete = false;
                        }
                        else
                        {
                            //if it is complete, we'll need to sort the data because subrequests may not come in in order
                            _arrivedHistoricalData[id] = _arrivedHistoricalData[id].OrderBy(x => x.DT).ToList();
                        }
                    }
                }

                if(requestComplete)
                    HistoricalDataRequestComplete(id);
            }
        }

        /// <summary>
        /// This method is called when a historical data request has delivered all its bars
        /// </summary>
        /// <param name="requestID"></param>
        private void HistoricalDataRequestComplete(int requestID)
        {
            //IB doesn't actually allow us to provide a deterministic starting point for our historical data query
            //so sometimes we get data from earlier than we want
            //here we throw it away
            var cutoffDate = _historicalDataRequests[requestID].StartingDate.Date;

            List<OHLCBar> bars;
            var removed = _arrivedHistoricalData.TryRemove(requestID, out bars);
            if (!removed) return;

            //due to the nature of sub-requests, the list may contain the same points multiple times
            //so we grab unique values only
            bars = bars.Distinct((x, y) => x.DT == y.DT).ToList();

            //if the data is daily or lower freq, and a stock, set adjusted ohlc values for convenience
            if (_historicalDataRequests[requestID].Frequency >= QDMS.BarSize.OneDay && 
                _historicalDataRequests[requestID].Instrument.Type == InstrumentType.Stock)
            {
                foreach (OHLCBar b in bars)
                {
                    b.AdjOpen = b.Open;
                    b.AdjHigh = b.High;
                    b.AdjLow = b.Low;
                    b.AdjClose = b.Close;
                }
            }

            RaiseEvent(HistoricalDataArrived, this, new QDMS.HistoricalDataEventArgs(
                _historicalDataRequests[requestID],
                bars.Where(x => x.DT.Date >= cutoffDate).ToList()));
        }

        //This event is raised when real time data arrives
        //We convert them and pass them on downstream
        private void _client_RealTimeBar(object sender, RealTimeBarEventArgs e)
        {
            RealTimeDataEventArgs args = TWSUtils.RealTimeDataEventArgsConverter(e);
            args.Symbol = _realTimeDataRequests[e.RequestId].Instrument.Symbol;
            RaiseEvent(DataReceived, this, args);
        }

        //This event is raised when the connection to TWS client closed.
        private void _client_ConnectionClosed(object sender, ConnectionClosedEventArgs e)
{'fraction_non_alphanumeric': 0.09107573059599874, 'fraction_numerical': 0.002932858489577878, 'mean_word_length': 1.7702016538517336, 'pattern_counts': {'":': 0, '<': 66, '<?xml version=': 0, '>': 82, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3684164', 'n_tokens_mistral': 4867, 'n_tokens_neox': 4541, 'n_words': 1482}