marketdata/MarketDataLib/Helper/InsiderTransactionMarketDataHelper.cs

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 Year { 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<String> 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<String>();
				symbols.Add(symbol);
				currentIndex = 0;
				while (true)
				{
					List<String> 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<String> symbols)
    {
			Profiler profiler=new Profiler();
			try
			{
				this.symbols = symbols;
				currentIndex = 0;
				while (true)
				{
					List<String> 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 LoadInsiderTransactionsYear(List<String> symbols,int year)
    {
			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<String>(UpdateManager.Entries)).ToList();

				while (true)
				{
					List<String> 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.Year=year;
						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<String> GetQueueSymbols()
    {
      List<String> queueSymbols = new List<String>();
      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.Year);
      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 year)
    {
      symbol = symbol.ToUpper();
      MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Load insider transactions for {0} years>={1}", symbol,year));
      InsiderTransactions insiderTransactions = MarketDataHelper.GetInsiderTransactionsYear(symbol, year);
      if (null == insiderTransactions || 0 == insiderTransactions.Count)
      {
        MDTrace.WriteLine(LogLevel.DEBUG,String.Format("No insider transactions for {0} years>={1}", symbol,year));
        return;
      }
      MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Insider Transactions, Saving {0} records for {1} years>={2}", insiderTransactions.Count, symbol, year));
      InsiderTransactionDA.DeleteInsiderTransactionsYearsGreaterEqual(symbol, year);
      InsiderTransactionDA.InsertInsiderTransactions(insiderTransactions);
      MDTrace.WriteLine(LogLevel.DEBUG,"Insider Transactions - Done.");
    }
  }
}