using System; using System.Collections.Generic; using System.Linq; using System.Threading; using MarketData.MarketDataModel; using MarketData.DataAccess; using MarketData.Utils; namespace MarketData.Helper { public class InsiderTransactionThreadHelper : ThreadHelper { public InsiderTransactionThreadHelper(String symbol, ManualResetEvent resetEvent) : base(symbol, resetEvent) { } public int YearGreaterEqual { get; set; } } public class InsiderTransactionMarketDataHelper { private static int MaxThreads = 10; // 10 threads avoids receiving HTTP Response 429 (Too many requests) private static int WAIT_TIME_MS=500; // wait between request private List symbols; private int currentIndex = 0; private UpdateManager UpdateManager = new UpdateManager(); public InsiderTransactionMarketDataHelper() { } public bool LoadInsiderTransactions(String symbol) { Profiler profiler=new Profiler(); try { symbols = new List(); symbols.Add(symbol); currentIndex = 0; while (true) { List queueSymbols = GetQueueSymbols(); if (null == queueSymbols || 0 == queueSymbols.Count) break; ManualResetEvent[] resetEvents = new ManualResetEvent[queueSymbols.Count]; for (int eventIndex = 0; eventIndex < resetEvents.Length; eventIndex++) { resetEvents[eventIndex] = new ManualResetEvent(false); } for (int index = 0; index < queueSymbols.Count; index++) { ThreadHelper threadHelper = new ThreadHelper(queueSymbols[index], resetEvents[index]); ThreadPool.QueueUserWorkItem(ThreadPoolCallback, threadHelper); } MDTrace.WriteLine(LogLevel.DEBUG,"Insider Transactions, waiting for queued items to complete."); WaitHandle.WaitAll(resetEvents); } return true; } finally { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LoadInsiderTransactions]End, total took {0}(ms)",profiler.End())); } } public bool LoadInsiderTransactions(List symbols) { Profiler profiler=new Profiler(); try { this.symbols = symbols; currentIndex = 0; while (true) { List queueSymbols = GetQueueSymbols(); if (null == queueSymbols || 0 == queueSymbols.Count) break; ManualResetEvent[] resetEvents = new ManualResetEvent[queueSymbols.Count]; for (int eventIndex = 0; eventIndex < resetEvents.Length; eventIndex++) { resetEvents[eventIndex] = new ManualResetEvent(false); } for (int index = 0; index < queueSymbols.Count; index++) { ThreadHelper threadHelper = new ThreadHelper(queueSymbols[index], resetEvents[index]); ThreadPool.QueueUserWorkItem(ThreadPoolCallback, threadHelper); try { Thread.Sleep(WAIT_TIME_MS); } catch(Exception) { ;} // wait } MDTrace.WriteLine(LogLevel.DEBUG,"Insider Transactions, waiting for queued items to complete."); WaitHandle.WaitAll(resetEvents); } return true; } finally { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LoadInsiderTransactions]End, total took {0}(ms)",profiler.End())); } } public bool LoadInsiderTransactionsYearGreaterEqual(List symbols,int yearGreaterEqual) { Profiler profiler=new Profiler(); try { this.symbols = symbols; currentIndex = 0; UpdateManager.Prepare("load_insider_transactions_year.txt", 7); // use max age 7 days this.symbols=this.symbols.Except(new List(UpdateManager.Entries)).ToList(); while (true) { List queueSymbols = GetQueueSymbols(); if (null == queueSymbols || 0 == queueSymbols.Count) break; ManualResetEvent[] resetEvents = new ManualResetEvent[queueSymbols.Count]; for (int eventIndex = 0; eventIndex < resetEvents.Length; eventIndex++) { resetEvents[eventIndex] = new ManualResetEvent(false); } for (int index = 0; index < queueSymbols.Count; index++) { InsiderTransactionThreadHelper threadHelper = new InsiderTransactionThreadHelper(queueSymbols[index], resetEvents[index]); threadHelper.UpdateManager=UpdateManager; threadHelper.YearGreaterEqual=yearGreaterEqual; ThreadPool.QueueUserWorkItem(ThreadPoolCallbackYearGreaterEqual, threadHelper); try { Thread.Sleep(WAIT_TIME_MS); } catch(Exception) { ;} // wait } MDTrace.WriteLine(LogLevel.DEBUG,"Insider Transactions, waiting for queued items to complete."); WaitHandle.WaitAll(resetEvents); } return true; } finally { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LoadInsiderTransactions]End, total took {0}(ms)",profiler.End())); } } private List GetQueueSymbols() { List queueSymbols = new List(); int index = currentIndex; for (; index < currentIndex + MaxThreads && index < symbols.Count; index++) { queueSymbols.Add(symbols[index]); } currentIndex = index; return queueSymbols; } public void ThreadPoolCallback(Object threadHelperContext) { ThreadHelper threadHelper = (ThreadHelper)threadHelperContext; MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Insider Transactions Thread {0} started for {1}...", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol)); LoadInsiderTransactionsSymbolEx(threadHelper.Symbol); MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Insider Transactions Thread {0} ended for {1}...", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol)); threadHelper.ResetEvent.Set(); } public static void LoadInsiderTransactionsSymbolEx(String symbol) { symbol = symbol.ToUpper(); MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Load insider transactions for {0}", symbol)); InsiderTransactions insiderTransactions = MarketDataHelper.GetInsiderTransactions(symbol); if (null == insiderTransactions || 0 == insiderTransactions.Count) { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("No insider transactions for {0}", symbol)); return; } MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Insider Transactions, Saving {0} records for {1}", insiderTransactions.Count, symbol)); InsiderTransactionDA.InsertInsiderTransactions(insiderTransactions); MDTrace.WriteLine(LogLevel.DEBUG,"Insider Transactions - Done."); } public void ThreadPoolCallbackYearGreaterEqual(Object threadHelperContext) { InsiderTransactionThreadHelper threadHelper = (InsiderTransactionThreadHelper)threadHelperContext; MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Insider Transactions Thread {0} started for {1}...", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol)); LoadInsiderTransactionsYearGreaterEqualEx(threadHelper.Symbol,threadHelper.YearGreaterEqual); MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Insider Transactions Thread {0} ended for {1}...", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol)); threadHelper.UpdateManager.Add(threadHelper.Symbol); threadHelper.ResetEvent.Set(); } public static void LoadInsiderTransactionsYearGreaterEqualEx(String symbol, int yearGreaterEqual) { symbol = symbol.ToUpper(); MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Load insider transactions for {0} years>={1}", symbol,yearGreaterEqual)); InsiderTransactions insiderTransactions = MarketDataHelper.GetInsiderTransactionsYear(symbol, yearGreaterEqual); if (null == insiderTransactions || 0 == insiderTransactions.Count) { MDTrace.WriteLine(LogLevel.DEBUG,String.Format("No insider transactions for {0} years>={1}", symbol,yearGreaterEqual)); return; } MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Insider Transactions, Saving {0} records for {1} years>={2}", insiderTransactions.Count, symbol, yearGreaterEqual)); InsiderTransactionDA.DeleteInsiderTransactionsYearsGreaterEqual(symbol, yearGreaterEqual); InsiderTransactionDA.InsertInsiderTransactions(insiderTransactions); MDTrace.WriteLine(LogLevel.DEBUG,"Insider Transactions - Done."); } } }