Sean/marketdata
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor GlobalPriceCache and LocalPriceCache to pattern snapshot, fetch, update. Parellelize.

Browse Source
This commit is contained in:
Sean
2026-02-24 12:23:38 -05:00
parent faded8cd55
commit bffff6c296
2 changed files with 521 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

403
MarketDataLib/Cache/GBPriceCache.cs
View File

@@ -1,260 +1,323 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Text;
using MarketData.MarketDataModel;
using MarketData.DataAccess;
using MarketData.Utils;
using System.Linq; using System.Linq;
using System.Threading;
using MarketData.MarketDataModel;
using MarketData.Utils;
using MarketData.Helper; using MarketData.Helper;
using MarketData.Numerical; using MarketData.Numerical;
using System.Threading; using MarketData.DataAccess;
// This cache is mainly used by the models. It is a short lived cache that gets cleared out every 2 minutes.
// This cache will attempt to load a price from the database if it is found in the cache.
namespace MarketData.Cache namespace MarketData.Cache
{ {
public class GBPriceCache public interface IPricingDataAccess
{ {
private Thread cacheMonitorThread=null; Price GetPrice(string symbol, DateTime date);
private volatile bool threadRun=true; Prices GetPrices(string symbol, DateTime maxDate, DateTime minDate);
private Object thisLock=new Object(); DateTime GetLatestDateOnOrBefore(string symbol, DateTime date);
private Dictionary<String,PricesByDate> priceCache=new Dictionary<String,PricesByDate>(); // the main cache }
private Dictionary<String,Price> realTimePriceCache=new Dictionary<String,Price>(); // short lived cache of realtime prices gets cleared out every cacheRefreshAfter(ms)
private Dictionary<String,bool> nullCache=new Dictionary<String,bool>(); internal class RealPricingDA : IPricingDataAccess
private DateGenerator dateGenerator=new DateGenerator(); {
private static GBPriceCache priceCacheInstance=null; public Price GetPrice(string symbol, DateTime date) => PricingDA.GetPrice(symbol, date);
private int cacheRefreshAfter=120000; // the cache will be cleaned up after 2 minutes public Prices GetPrices(string symbol, DateTime maxDate, DateTime minDate) => PricingDA.GetPrices(symbol, maxDate, minDate);
public DateTime GetLatestDateOnOrBefore(string symbol, DateTime date) => PricingDA.GetLatestDateOnOrBefore(symbol, date);
}
internal class CacheSnapshot
{
public Dictionary<String, PricesByDate> PriceCache { get; }
public Dictionary<String, Price> RealTimePriceCache { get; }
public Dictionary<String, bool> NullCache { get; }
public CacheSnapshot(
Dictionary<String, PricesByDate> priceCache,
Dictionary<String, Price> realTimePriceCache,
Dictionary<String, bool> nullCache)
{
PriceCache = priceCache;
RealTimePriceCache = realTimePriceCache;
NullCache = nullCache;
}
}
public class GBPriceCache : IDisposable
{
private Thread cacheMonitorThread = null;
private volatile bool threadRun = true;
private Object thisLock = new Object();
private CacheSnapshot snapshot;
private DateGenerator dateGenerator = new DateGenerator();
private static GBPriceCache priceCacheInstance = null;
private int cacheRefreshAfter = 120000; // 2 minutes
private SemaphoreSlim fetchSemaphore = new SemaphoreSlim(8); // max 8 concurrent DB fetches
public IPricingDataAccess PricingDataAccess { get; set; } = new RealPricingDA();
protected GBPriceCache() protected GBPriceCache()
{ {
cacheMonitorThread=new Thread(new ThreadStart(ThreadProc)); snapshot = new CacheSnapshot(new Dictionary<String, PricesByDate>(), new Dictionary<String, Price>(), new Dictionary<String, bool>());
cacheMonitorThread = new Thread(new ThreadStart(ThreadProc));
cacheMonitorThread.Start(); cacheMonitorThread.Start();
} }
public static GBPriceCache GetInstance() public static GBPriceCache GetInstance()
{ {
lock(typeof(GBPriceCache)) lock (typeof(GBPriceCache))
{ {
if(null==priceCacheInstance) if (null == priceCacheInstance)
{ {
priceCacheInstance=new GBPriceCache(); priceCacheInstance = new GBPriceCache();
} }
return priceCacheInstance; return priceCacheInstance;
} }
} }
public void Clear() public void Clear()
{ {
lock(thisLock) lock (thisLock)
{ {
priceCache=new Dictionary<String,PricesByDate>(); snapshot = new CacheSnapshot(new Dictionary<String, PricesByDate>(), new Dictionary<String, Price>(), new Dictionary<String, bool>());
realTimePriceCache=new Dictionary<String,Price>();
nullCache=new Dictionary<String,bool>();
}
}
public void Dispose()
{
lock(thisLock)
{
if(null==priceCacheInstance || false==threadRun)return;
threadRun=false;
if(null!=cacheMonitorThread)
{
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[GBPriceCache:Dispose]Thread state is '{0}'. Joining main thread...",Utility.ThreadStateToString(cacheMonitorThread)));
cacheMonitorThread.Join(5000);
this.cacheMonitorThread=null;
}
MDTrace.WriteLine(LogLevel.DEBUG,"[GBPriceCache:Dispose] End.");
priceCacheInstance=null;
}
}
public void ClearCacheOnOrBefore(DateTime onOrBeforeDate,bool collect=false)
{
lock(thisLock)
{
MDTrace.WriteLine(LogLevel.DEBUG,"Clearing GBPriceCache cache.");
List<String> symbols=new List<String>(priceCache.Keys);
foreach(String symbol in symbols)
{
PricesByDate pricesByDate=priceCache[symbol];
List<DateTime> symbolDates=new List<DateTime>(pricesByDate.Keys);
foreach(DateTime symbolDate in symbolDates)
{
if(symbolDate<onOrBeforeDate) pricesByDate.Remove(symbolDate);
}
}
MDTrace.WriteLine(LogLevel.DEBUG,"Calling garbage collector...");
if(collect) GC.Collect();
} }
} }
public Price GetPriceOrLatestAvailable(String symbol,DateTime date) public void Dispose()
{ {
lock(thisLock) lock (thisLock)
{ {
Price price=GetPrice(symbol,date); if (null == priceCacheInstance || !threadRun) return;
if(null!=price) return price; threadRun = false;
DateTime latestPricingDate=PricingDA.GetLatestDateOnOrBefore(symbol,date); if (null != cacheMonitorThread)
price=GetPrice(symbol,latestPricingDate); {
if(null!=price) return price; MDTrace.WriteLine(LogLevel.DEBUG, "[GBPriceCache:Dispose] Joining monitor thread...");
price=PricingDA.GetPrice(symbol,latestPricingDate); cacheMonitorThread.Join(5000);
if(null!=price) AddPrice(price); cacheMonitorThread = null;
}
priceCacheInstance = null;
}
}
public void ClearCacheOnOrBefore(DateTime onOrBeforeDate, bool collect = false)
{
lock (thisLock)
{
Dictionary<String, PricesByDate> newPriceCache = new Dictionary<String, PricesByDate>();
foreach (KeyValuePair<String, PricesByDate> entry in snapshot.PriceCache)
{
String symbol = entry.Key;
PricesByDate filteredPrices = new PricesByDate();
PricesByDate existingPrices = entry.Value;
foreach (KeyValuePair<DateTime, Price> kvp in existingPrices)
{
if (kvp.Key >= onOrBeforeDate)
{
filteredPrices.Add(kvp.Key, kvp.Value);
}
}
if (filteredPrices.Count > 0)
{
newPriceCache.Add(symbol, filteredPrices);
}
}
UpdateSnapshot(newPriceCache, snapshot.RealTimePriceCache, snapshot.NullCache);
if (collect) GC.Collect();
}
}
public Price GetPriceOrLatestAvailable(String symbol, DateTime date)
{
Price price = GetPrice(symbol, date);
if (null != price) return price;
DateTime latestPricingDate = PricingDataAccess.GetLatestDateOnOrBefore(symbol, date);
price = GetPrice(symbol, latestPricingDate);
if (null != price) return price;
fetchSemaphore.Wait();
try
{
price = PricingDataAccess.GetPrice(symbol, latestPricingDate);
}
finally
{
fetchSemaphore.Release();
}
if (null !=price) AddPrice(price);
return price; return price;
} }
}
public Price GetRealtimePrice(String symbol) public Price GetRealtimePrice(String symbol)
{ {
if(realTimePriceCache.ContainsKey(symbol)) return realTimePriceCache[symbol]; if (snapshot.RealTimePriceCache.ContainsKey(symbol))
Price price=MarketDataHelper.GetLatestPrice(symbol);
if(null!=price)
{ {
realTimePriceCache.Add(symbol,price); return snapshot.RealTimePriceCache[symbol];
}
Price price = MarketDataHelper.GetLatestPrice(symbol);
if (null != price)
{
Dictionary<String, Price> newRealtime = new Dictionary<String, Price>(snapshot.RealTimePriceCache);
newRealtime.Add(symbol, price);
UpdateSnapshot(snapshot.PriceCache, newRealtime, snapshot.NullCache);
} }
return price; return price;
} }
public Price GetPrice(String symbol,DateTime date)
public Price GetPrice(String symbol, DateTime date)
{ {
lock(thisLock) date = date.Date;
if (!ContainsPrice(symbol, date))
{ {
date=date.Date; String key = symbol + Utility.DateTimeToStringMMHDDHYYYY(date);
if(!ContainsPrice(symbol,date)) if (snapshot.NullCache.ContainsKey(key))
{ {
String key=symbol+Utility.DateTimeToStringMMHDDHYYYY(date); return null;
if(nullCache.ContainsKey(key)) return null;
Price price=PricingDA.GetPrice(symbol,date);
if(null==price)
{
nullCache.Add(key,true);
return price;
} }
fetchSemaphore.Wait();
Price price;
try
{
price = PricingDataAccess.GetPrice(symbol, date);
}
finally
{
fetchSemaphore.Release();
}
if (null ==price)
{
Dictionary<String, bool> newNullCache = new Dictionary<String, bool>(snapshot.NullCache);
newNullCache.Add(key, true);
UpdateSnapshot(snapshot.PriceCache, snapshot.RealTimePriceCache, newNullCache);
return null;
}
AddPrice(price); AddPrice(price);
} }
if(!priceCache.ContainsKey(symbol)) return null; if (!snapshot.PriceCache.ContainsKey(symbol)) return null;
PricesByDate pricesByDate=priceCache[symbol]; PricesByDate pricesByDate = snapshot.PriceCache[symbol];
if(!pricesByDate.ContainsKey(date.Date)) return null; if (!pricesByDate.ContainsKey(date)) return null;
return pricesByDate[date]; return pricesByDate[date];
} }
}
public Prices GetPrices(String symbol, DateTime earlierDate, DateTime laterDate) public Prices GetPrices(String symbol, DateTime earlierDate, DateTime laterDate)
{ {
DateGenerator dateGenerator = new DateGenerator(); DateGenerator localDateGenerator = new DateGenerator();
if (laterDate < earlierDate)
if(laterDate<earlierDate)
{ {
DateTime tempDate = earlierDate; DateTime tempDate = earlierDate;
earlierDate = laterDate; earlierDate = laterDate;
laterDate=tempDate; laterDate = tempDate;
} }
List<DateTime> datesList = dateGenerator.GenerateHistoricalDates(earlierDate, laterDate); List<DateTime> datesList = localDateGenerator.GenerateHistoricalDates(earlierDate, laterDate);
datesList = datesList.Where(x => x >= earlierDate).ToList(); datesList = datesList.Where(x => x >= earlierDate).ToList();
return GetPrices(symbol, laterDate, datesList.Count); return GetPrices(symbol, laterDate, datesList.Count);
} }
// The most recent price is returned at the lowest index public Prices GetPrices(String symbol, DateTime startDate, int dayCount)
public Prices GetPrices(String symbol,DateTime startDate,int dayCount)
{ {
lock(thisLock) List<DateTime> historicalDates = dateGenerator.GenerateHistoricalDates(startDate, dayCount + 60);
List<DateTime> missingDates = new List<DateTime>();
foreach (DateTime historicalDate in historicalDates)
{ {
List<DateTime> historicalDates=dateGenerator.GenerateHistoricalDates(startDate,dayCount+60); if (!ContainsPrice(symbol, historicalDate))
Prices prices=null;
List<DateTime> missingDates=null;
foreach(DateTime historicalDate in historicalDates)
{ {
if(!ContainsPrice(symbol,historicalDate)) String key = symbol + Utility.DateTimeToStringMMHDDHYYYY(historicalDate);
if (!snapshot.NullCache.ContainsKey(key))
{ {
String key=symbol+Utility.DateTimeToStringMMHDDHYYYY(historicalDate);
if(nullCache.ContainsKey(key)) continue;
if(null==missingDates)missingDates=new List<DateTime>();
missingDates.Add(historicalDate); missingDates.Add(historicalDate);
} }
} }
if(null!=missingDates)
{
DateTime minDate=(from DateTime date in missingDates select date).Min();
DateTime maxDate=(from DateTime date in missingDates select date).Max();
prices=PricingDA.GetPrices(symbol,maxDate,minDate);
foreach(Price price in prices) AddPrice(price);
prices=new Prices();
foreach(DateTime historicalDate in historicalDates)
{
if(!ContainsPrice(symbol,historicalDate))
{
String key=symbol+Utility.DateTimeToStringMMHDDHYYYY(historicalDate);
if(!nullCache.ContainsKey(key)) nullCache.Add(key,true);
} }
else if (missingDates.Count > 0)
{ {
if(!priceCache.ContainsKey(symbol)) continue; DateTime minDate = missingDates.Min();
PricesByDate pricesByDate=priceCache[symbol]; DateTime maxDate = missingDates.Max();
if(!pricesByDate.ContainsKey(historicalDate.Date)) continue;
fetchSemaphore.Wait();
Prices loadedPrices;
try
{
loadedPrices = PricingDataAccess.GetPrices(symbol, maxDate, minDate);
}
finally
{
fetchSemaphore.Release();
}
foreach (Price price in loadedPrices)
{
AddPrice(price);
}
}
Prices prices = new Prices();
foreach (DateTime historicalDate in historicalDates)
{
if (!snapshot.PriceCache.ContainsKey(symbol)) continue;
PricesByDate pricesByDate = snapshot.PriceCache[symbol];
if (!pricesByDate.ContainsKey(historicalDate)) continue;
prices.Add(pricesByDate[historicalDate]); prices.Add(pricesByDate[historicalDate]);
} }
List<Price> ordered = prices.OrderByDescending(x => x.Date).ToList();
return new Prices(ordered.Take(dayCount).ToList());
} }
}
else
{
prices=new Prices();
foreach(DateTime historicalDate in historicalDates)
{
if(!priceCache.ContainsKey(symbol)) continue;
if(!priceCache[symbol].ContainsKey(historicalDate.Date))
{
continue;
}
prices.Add((priceCache[symbol])[historicalDate]);
}
}
return new Prices(prices.OrderByDescending(x => x.Date).ToList().Take(dayCount).ToList());
}
}
private void AddPrice(Price price) private void AddPrice(Price price)
{ {
lock(thisLock) if (null == price) return;
lock (thisLock)
{ {
if(null==price) return; PricesByDate pricesByDate;
PricesByDate pricesByDate=null; if (!snapshot.PriceCache.ContainsKey(price.Symbol))
if(!priceCache.ContainsKey(price.Symbol))
{ {
pricesByDate=new PricesByDate(); pricesByDate = new PricesByDate();
pricesByDate.Add(price.Date,price); pricesByDate.Add(price.Date, price);
priceCache.Add(price.Symbol,pricesByDate); Dictionary<String, PricesByDate> newCache = new Dictionary<String, PricesByDate>(snapshot.PriceCache);
newCache.Add(price.Symbol, pricesByDate);
UpdateSnapshot(newCache, snapshot.RealTimePriceCache, snapshot.NullCache);
} }
else else
{ {
pricesByDate=priceCache[price.Symbol]; pricesByDate = snapshot.PriceCache[price.Symbol];
if(pricesByDate.ContainsKey(price.Date.Date)) return; if (!pricesByDate.ContainsKey(price.Date))
pricesByDate.Add(price.Date.Date,price);
}
}
}
public bool ContainsPrice(String symbol,DateTime date)
{ {
if(!priceCache.ContainsKey(symbol)) return false; pricesByDate.Add(price.Date, price);
PricesByDate pricesByDate=priceCache[symbol];
if(!pricesByDate.ContainsKey(date.Date)) return false;
return true;
} }
}
}
}
public bool ContainsPrice(String symbol, DateTime date)
{
if (!snapshot.PriceCache.ContainsKey(symbol)) return false;
PricesByDate pricesByDate = snapshot.PriceCache[symbol];
return pricesByDate.ContainsKey(date);
}
private void ThreadProc() private void ThreadProc()
{ {
int quantums=0; int quantums = 0;
int quantumInterval=1000; int quantumInterval = 1000;
while(threadRun)
while (threadRun)
{ {
Thread.Sleep(quantumInterval); Thread.Sleep(quantumInterval);
if(!threadRun) break; if(!threadRun)break;
quantums+=quantumInterval; quantums += quantumInterval;
if(quantums>cacheRefreshAfter) if (quantums > cacheRefreshAfter)
{ {
quantums=0; quantums = 0;
lock(thisLock) lock (thisLock)
{ {
realTimePriceCache.Clear(); UpdateSnapshot(snapshot.PriceCache, new Dictionary<String, Price>(), snapshot.NullCache);
MDTrace.WriteLine(LogLevel.DEBUG,"Clearing GBPriceCache price cache.");
} }
} }
} }
MDTrace.WriteLine(LogLevel.DEBUG,"[GBPriceCache:ThreadProc]Thread ended."); }
private void UpdateSnapshot(Dictionary<String, PricesByDate> newPriceCache,Dictionary<String, Price> newRealtimePriceCache, Dictionary<String, bool> newNullCache)
{
snapshot = new CacheSnapshot(newPriceCache, newRealtimePriceCache, newNullCache);
} }
} }
} }

411
MarketDataLib/Cache/LocalPriceCache.cs
View File

@@ -1,65 +1,70 @@
using System; using MarketData.MarketDataModel;
using System.Linq;
using System.Collections.Generic;
using MarketData.MarketDataModel;
using MarketData.Utils; using MarketData.Utils;
using MarketData.DataAccess; using MarketData.DataAccess;
using System.Collections.Concurrent;
using System.Threading; using System.Threading;
using System.Collections.Generic;
using System;
using System.Threading.Tasks;
using System.Linq;
// This cache is mainly used by gainloss generator. This cache is intended to be front loaded and then used.
// This cache will not attempt to load an item that is not found. It does have a Refresh() that will reload only the most recent pricing data from the database in order to
// maintain the most updated pricing.
namespace MarketData.Cache namespace MarketData.Cache
{ {
public class LocalPriceCache public class LocalPriceCache
{ {
private Dictionary<String,PricesByDate> priceCache=new Dictionary<String,PricesByDate>(); private Dictionary<string, PricesByDate> priceCache = new Dictionary<string, PricesByDate>();
private static LocalPriceCache instance=null; private static LocalPriceCache instance = null;
private DateTime latestDate = Utility.Epoch; private DateTime latestDate = Utility.Epoch;
private Thread cacheMonitorThread=null; private Thread cacheMonitorThread = null;
private volatile bool threadRun=true; private volatile bool threadRun = true;
private int cacheCycle=300000; private int cacheCycle = 300000;
private Object thisLock=new Object(); private object thisLock = new object();
private object fetchLock = new object();
private LocalPriceCache() private LocalPriceCache()
{ {
cacheMonitorThread=new Thread(new ThreadStart(ThreadProc)); cacheMonitorThread = new Thread(new ThreadStart(ThreadProc));
cacheMonitorThread.Start(); cacheMonitorThread.Start();
} }
public void Clear() public void Clear()
{ {
lock(thisLock) lock (thisLock)
{ {
priceCache=new Dictionary<String,PricesByDate>(); priceCache = new Dictionary<string, PricesByDate>();
RefreshLatestDate(); RefreshLatestDate();
} }
} }
public void Dispose() public void Dispose()
{ {
lock(thisLock) Thread threadToJoin = null;
lock (thisLock)
{ {
if(null==instance || false==threadRun)return; if (instance == null || !threadRun) return;
threadRun=false; threadRun = false;
if(null!=cacheMonitorThread) threadToJoin = cacheMonitorThread;
cacheMonitorThread = null;
instance = null;
}
if (threadToJoin != null)
{ {
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LocalPriceCache:Dispose]Thread state is '{0}'. Joining main thread...",Utility.ThreadStateToString(cacheMonitorThread))); MDTrace.WriteLine(LogLevel.DEBUG, $"[LocalPriceCache:Dispose] Thread state is '{Utility.ThreadStateToString(threadToJoin)}'. Joining...");
cacheMonitorThread.Join(5000); threadToJoin.Join(5000);
this.cacheMonitorThread=null;
}
MDTrace.WriteLine(LogLevel.DEBUG,"[LocalPriceCache:Dispose] End");
instance=null;
} }
MDTrace.WriteLine(LogLevel.DEBUG, "[LocalPriceCache:Dispose] End");
} }
public static LocalPriceCache GetInstance() public static LocalPriceCache GetInstance()
{ {
lock(typeof(LocalPriceCache)) lock (typeof(LocalPriceCache))
{ {
if(null==instance) if (instance == null)
{ {
instance=new LocalPriceCache(); instance = new LocalPriceCache();
} }
return instance; return instance;
} }
@@ -67,17 +72,17 @@ namespace MarketData.Cache
public void RefreshLatestDate() public void RefreshLatestDate()
{ {
lock(typeof(LocalPriceCache)) lock (thisLock)
{ {
latestDate=PricingDA.GetLatestDate(); latestDate = PricingDA.GetLatestDate();
} }
} }
public DateTime GetLatestDate() public DateTime GetLatestDate()
{ {
lock(typeof(LocalPriceCache)) lock (thisLock)
{ {
if(Utility.IsEpoch(latestDate)) if (Utility.IsEpoch(latestDate))
{ {
RefreshLatestDate(); RefreshLatestDate();
} }
@@ -87,65 +92,118 @@ namespace MarketData.Cache
public void Refresh() public void Refresh()
{ {
lock(typeof(LocalPriceCache)) List<string> symbols;
Dictionary<string, DateTime> currentMaxDates;
lock (thisLock)
{ {
List<String> symbols=new List<String>(priceCache.Keys); symbols = priceCache.Keys.ToList();
Dictionary<String, DateTime> maxDbDates = PricingDA.GetLatestDates(symbols); currentMaxDates = priceCache.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.MaxDate);
RefreshLatestDate(); }
foreach(String symbol in symbols)
if (symbols.Count == 0) return;
ConcurrentDictionary<string, PricesByDate> fullReloads = new ConcurrentDictionary<string, PricesByDate>();
ConcurrentDictionary<string, Price> singleUpdates = new ConcurrentDictionary<string, Price>();
DateTime latestDateFromDb;
lock (fetchLock)
{ {
PricesByDate symbolPrices=priceCache[symbol]; Dictionary<string, DateTime> maxDbDates = PricingDA.GetLatestDates(symbols);
DateTime maxDate=symbolPrices.MaxDate; // get the latest date in the cache latestDateFromDb = PricingDA.GetLatestDate();
if(maxDbDates.ContainsKey(symbol) && !maxDbDates[symbol].Date.Equals(maxDate.Date)) // if the cache date and the database date are not equal then reload the cache
Parallel.ForEach(symbols, new ParallelOptions { MaxDegreeOfParallelism = 8 }, symbol =>
{ {
MDTrace.WriteLine(LogLevel.DEBUG,$"Cache date and Database date for {symbol} are not equal, reloading cache. Cache Date:{maxDate.ToShortDateString()} Database Date:{maxDbDates[symbol].Date.ToShortDateString()}"); if (!currentMaxDates.TryGetValue(symbol, out var cachedMax)) return;
Prices prices=PricingDA.GetPrices(symbol,symbolPrices.MinDate); // reload the prices for this symbol using the current minDate in the cache as a lower boundary
if(null==prices)continue; // if we can't load any prices for symbol then just continue if (maxDbDates.TryGetValue(symbol, out var dbMax) && dbMax.Date != cachedMax.Date)
priceCache.Remove(symbol); // remove the pricing entries in the price cache for the symbol {
priceCache.Add(symbol,prices.GetPricesByDate()); // reload the cache Prices prices = PricingDA.GetPrices(symbol, cachedMax);
if (prices != null) fullReloads[symbol] = prices.GetPricesByDate();
} }
else else
{ {
MDTrace.WriteLine(LogLevel.DEBUG,$"[LocalPriceCache] Fetching latest price from database for {symbol} on {maxDate.ToShortDateString()}"); Price price = PricingDA.GetPrice(symbol, cachedMax);
Price price=PricingDA.GetPrice(symbol,maxDate); // the max date from the cache equals the max date from the database so just reload the latest price from the database if (price != null) singleUpdates[symbol] = price;
if(null==price)continue; // if no latest price then just continue
symbolPrices.Remove(maxDate); // remove the current price associated with the max date
symbolPrices.Add(maxDate,price); // reload the latest price for maxDate(symbol) into the cache
} }
});
}
lock (thisLock)
{
latestDate = latestDateFromDb;
foreach (var kvp in fullReloads)
{
if (priceCache.TryGetValue(kvp.Key, out PricesByDate existing) && existing.MaxDate == currentMaxDates[kvp.Key])
{
priceCache[kvp.Key] = kvp.Value;
}
}
foreach (var kvp in singleUpdates)
{
if (priceCache.TryGetValue(kvp.Key, out PricesByDate pricesByDate) && pricesByDate.MaxDate == currentMaxDates[kvp.Key])
{
// Remove the old price (if any) and add the new price properly
if (pricesByDate.ContainsKey(kvp.Value.Date))
pricesByDate.Remove(kvp.Value.Date);
pricesByDate.Add(kvp.Value.Date, kvp.Value);
} }
} }
} }
// This version of Add(PortfolioTrades) will account for adding multiple lots at different times. So instead of just checking for the existance of the symbol in the cache MDTrace.WriteLine(LogLevel.DEBUG, $"Full reloads: {fullReloads.Count}, Single updates: {singleUpdates.Count}");
// we look to see if the symbol is in the cache and what dates are available. If the date range specified in the trade are not available then we load those date ranges. }
// This is a brute force approach always maintaining the gap between successive TradeDates in th.e portfolio trades and the maximum date for the symbol in the database.
// So while it is inefficient in terms of memory usage it alleviates the need for figuring out contiguous price sections
public void Add(PortfolioTrades portfolioTrades) public void Add(PortfolioTrades portfolioTrades)
{ {
lock(typeof(LocalPriceCache)) List<string> symbols = portfolioTrades.Symbols;
Dictionary<string, DateTime> minTradeDates = symbols.ToDictionary(sym => sym, sym => portfolioTrades.GetMinTradeDate(sym));
Dictionary<string, DateTime> minCacheDates;
lock (thisLock)
{ {
Profiler profiler=new Profiler(); minCacheDates = symbols.ToDictionary(sym => sym, sym => priceCache.ContainsKey(sym) ? priceCache[sym].MinDate : DateTime.MaxValue);
profiler.Start();
List<String> symbols=portfolioTrades.Symbols;
foreach(String symbol in symbols)
{
DateTime minPortfolioTradeDate=portfolioTrades.GetMinTradeDate(symbol);
if(!ContainsSymbol(symbol))
{
Prices prices=PricingDA.GetPrices(symbol,minPortfolioTradeDate);
if(null==prices)continue;
foreach(Price price in prices)Add(price);
} }
else
ConcurrentDictionary<string, Prices> fetchedPrices = new ConcurrentDictionary<string, Prices>();
Parallel.ForEach(symbols, new ParallelOptions { MaxDegreeOfParallelism = 8 }, symbol =>
{ {
DateTime minCacheDate=GetMinCacheDate(symbol); DateTime minTradeDate = minTradeDates[symbol];
if(minPortfolioTradeDate<minCacheDate) DateTime minCacheDate = minCacheDates[symbol];
Prices prices = null;
try
{ {
Prices prices=PricingDA.GetPrices(symbol,minCacheDate,minPortfolioTradeDate); // Fill the gap by retrieving prices starting at minCache date and going back in time to minPortfolioTradeDate if (minCacheDate == DateTime.MaxValue)
if(null==prices)continue; {
foreach(Price price in prices)Add(price); prices = PricingDA.GetPrices(symbol, minTradeDate);
PricesByDate p=priceCache[symbol];
} }
else if (minTradeDate < minCacheDate)
{
prices = PricingDA.GetPrices(symbol, minCacheDate, minTradeDate);
}
if (prices != null && prices.Count > 0)
{
fetchedPrices[symbol] = prices;
}
}
catch (Exception ex)
{
MDTrace.WriteLine(LogLevel.DEBUG, $"Error fetching prices for {symbol}: {ex.Message}");
}
});
lock (thisLock)
{
foreach (var kvp in fetchedPrices)
{
foreach (var price in kvp.Value)
{
Add(price);
} }
} }
} }
@@ -153,131 +211,186 @@ namespace MarketData.Cache
public void Add(Prices prices) public void Add(Prices prices)
{ {
foreach(Price price in prices) foreach (Price price in prices)
{ {
Add(price); Add(price);
} }
} }
public void Add(List<String> symbols,DateTime pricingDate) public void Add(List<string> symbols, DateTime pricingDate)
{ {
foreach(String symbol in symbols) if (symbols == null || symbols.Count == 0) return;
ConcurrentDictionary<string, Price> fetchedPrices = new ConcurrentDictionary<string, Price>();
Parallel.ForEach(symbols, new ParallelOptions { MaxDegreeOfParallelism = 8 }, symbol =>
{ {
if(ContainsPrice(symbol,pricingDate))continue; lock (thisLock)
Price price=PricingDA.GetPrice(symbol,pricingDate); {
if(null==price)continue; if (ContainsPrice(symbol, pricingDate)) return;
Add(price); }
try
{
Price price = PricingDA.GetPrice(symbol, pricingDate);
if (price != null) fetchedPrices[symbol] = price;
}
catch (Exception ex)
{
MDTrace.WriteLine(LogLevel.DEBUG, $"Error fetching price for {symbol} on {pricingDate:yyyy-MM-dd}: {ex.Message}");
}
});
lock (thisLock)
{
foreach (var kvp in fetchedPrices)
{
Add(kvp.Value);
}
} }
} }
public void Add(Price price) public void Add(Price price)
{ {
lock(typeof(LocalPriceCache)) if (price == null) return;
lock (thisLock)
{ {
if(null==price)return; if (!priceCache.TryGetValue(price.Symbol, out var pricesByDate))
if(ContainsPrice(price.Symbol,price.Date))return;
PricesByDate pricesByDate=null;
if(!priceCache.ContainsKey(price.Symbol))
{ {
pricesByDate=new PricesByDate(); pricesByDate = new PricesByDate();
pricesByDate.Add(price.Date,price); priceCache[price.Symbol] = pricesByDate;
priceCache.Add(price.Symbol,pricesByDate);
return;
} }
pricesByDate=priceCache[price.Symbol]; if (!pricesByDate.ContainsKey(price.Date))
if(pricesByDate.ContainsKey(price.Date))return; {
pricesByDate.Add(price.Date,price); pricesByDate.Add(price.Date, price); // must use Add() to update MinDate/MaxDate
} }
} }
public DateTime GetMinCacheDate(String symbol) }
public DateTime GetMinCacheDate(string symbol)
{ {
if(!ContainsSymbol(symbol))return Utility.Epoch; lock (thisLock)
PricesByDate symbolPrices=priceCache[symbol]; {
if (!priceCache.TryGetValue(symbol, out var symbolPrices) || symbolPrices.Count == 0)
{
return Utility.Epoch;
}
return symbolPrices.MinDate; return symbolPrices.MinDate;
} }
}
public void RemoveDate(DateTime date) public void RemoveDate(DateTime date)
{ {
lock(typeof(LocalPriceCache)) lock (thisLock)
{ {
List<String> symbols=new List<String>(priceCache.Keys); foreach (var kvp in priceCache)
foreach(String key in symbols)
{ {
PricesByDate pricesByDate=priceCache[key]; kvp.Value.Remove(date);
if(pricesByDate.ContainsKey(date))pricesByDate.Remove(date);
} }
}
}
public Price GetPrice(String symbol,DateTime date)
{
lock(typeof(LocalPriceCache))
{
if(!priceCache.ContainsKey(symbol))return null;
PricesByDate pricesByDate=priceCache[symbol];
if(!pricesByDate.ContainsKey(date))return null;
return pricesByDate[date];
}
}
public bool ContainsPrice(String symbol,DateTime date)
{
lock(typeof(LocalPriceCache))
{
if(!priceCache.ContainsKey(symbol))return false;
PricesByDate pricesByDate=priceCache[symbol];
if(!pricesByDate.ContainsKey(date))return false;
return true;
}
}
public bool ContainsPrice(List<String> symbols,DateTime date)
{
lock(typeof(LocalPriceCache))
{
foreach(String symbol in symbols)if(!ContainsPrice(symbol,date))return false;
return true;
} }
} }
public bool ContainsSymbol(String symbol) public Prices GetPrices(string symbol, DateTime endDate, int dayCount)
{ {
lock(typeof(LocalPriceCache)) lock (thisLock)
{
if (!priceCache.TryGetValue(symbol, out var pricesByDate)) return new Prices();
DateGenerator dateGenerator = new DateGenerator();
List<DateTime> historicalDates = dateGenerator.GenerateHistoricalDates(endDate, dayCount);
Prices result = new Prices();
foreach (DateTime date in historicalDates)
{
if (pricesByDate.ContainsKey(date))
{
result.Add(pricesByDate[date]);
}
}
return result;
}
}
public Price GetPrice(string symbol, DateTime date)
{
lock (thisLock)
{
if (!priceCache.TryGetValue(symbol, out var pricesByDate)) return null;
return pricesByDate.TryGetValue(date, out var price) ? price : null;
}
}
public bool ContainsPrice(string symbol, DateTime date)
{
lock (thisLock)
{
if (!priceCache.TryGetValue(symbol, out var pricesByDate)) return false;
return pricesByDate.ContainsKey(date);
}
}
public bool ContainsPrice(List<string> symbols, DateTime date)
{
if (symbols == null || symbols.Count == 0) return false;
lock (thisLock)
{
foreach (string symbol in symbols)
{
if (!priceCache.TryGetValue(symbol, out var pricesByDate) || !pricesByDate.ContainsKey(date))
{ {
if(priceCache.ContainsKey(symbol))return true;
return false; return false;
} }
} }
return true;
}
}
public long Count() public bool ContainsSymbol(string symbol)
{ {
long count=0; lock (thisLock)
List<String> symbols=priceCache.Keys.ToList();
foreach(String symbol in symbols)
{ {
PricesByDate pricesByDate=priceCache[symbol]; return priceCache.ContainsKey(symbol);
count+=pricesByDate.Count; }
}
private long Count()
{
lock (thisLock)
{
long count = 0;
foreach (var pricesByDate in priceCache.Values)
{
count += pricesByDate.Count;
} }
return count; return count;
} }
}
private void ThreadProc() private void ThreadProc()
{ {
int quantums=0; int quantums = 0;
int quantumInterval=1000; int quantumInterval = 1000;
long lastCount=0; long lastCount = 0;
while(threadRun)
while (threadRun)
{ {
Thread.Sleep(quantumInterval); Thread.Sleep(quantumInterval);
quantums+=quantumInterval; quantums += quantumInterval;
if(quantums>cacheCycle) if (quantums > cacheCycle)
{ {
quantums=0; quantums = 0;
lock(thisLock) lock (thisLock)
{ {
lastCount=Count(); lastCount = Count();
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LocalPriceCache:ThreadProc] Symbols: {0}. Items in cache: {1}.",priceCache.Keys.Count,Utility.FormatNumber(lastCount,0,true))); MDTrace.WriteLine(LogLevel.DEBUG, $"[LocalPriceCache:ThreadProc] Symbols: {priceCache.Keys.Count}. Items in cache: {Utility.FormatNumber(lastCount,0,true)}.");
} }
} }
} }
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LocalPriceCache:ThreadProc] Thread ended. Items in cache:{0}",Utility.FormatNumber(lastCount,0,true)));
MDTrace.WriteLine(LogLevel.DEBUG, $"[LocalPriceCache:ThreadProc] Thread ended. Items in cache:{Utility.FormatNumber(lastCount,0,true)}");
} }
} }
} }