marketdata/MarketDataLib/Helper/ETFHoldingsMarketDataHelper.cs

using System;
using System.Collections.Generic;
using System.Threading;
using MarketData.MarketDataModel;
using MarketData.DataAccess;
using MarketData.Utils;

namespace MarketData.Helper
{
  public class ETFHoldingsMarketDataHelper
  {
    private static int MaxThreads = 10; //  5 (int)ThreadHelperEnum.MaxThreads;
    private List<String> symbols;
    private int currentIndex = 0;
    private int WAIT_BETWEEN_REQUESTS=250;    // 500

    public ETFHoldingsMarketDataHelper()
    {
    }
		public bool LoadETFHoldings(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(ThreadPoolCallbackLoadETFHoldings, threadHelper);
            try{Thread.Sleep(WAIT_BETWEEN_REQUESTS);}catch(Exception){;}
					}
					MDTrace.WriteLine(LogLevel.DEBUG,"Load ETF Holdings, waiting for queued items to complete.");
					WaitHandle.WaitAll(resetEvents);
				}
				return true;
			}
			finally
			{
				MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LoadETFHoldings]End, total took {0}(ms)",profiler.End()));
			}
		}
		public bool LoadETFHoldings()
		{
			Profiler profiler=new Profiler();
			try
			{
				symbols =  PricingDA.GetSymbolsETF();
				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(ThreadPoolCallbackLoadETFHoldings, threadHelper);
            try{Thread.Sleep(WAIT_BETWEEN_REQUESTS);}catch(Exception){;}
					}
					MDTrace.WriteLine(LogLevel.DEBUG,"Load ETF Holdings, waiting for queued items to complete.");
					WaitHandle.WaitAll(resetEvents);
				}
				return true;
			}
			finally
			{
				MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[LoadETFHoldings]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 ThreadPoolCallbackLoadETFHoldings(Object threadHelperContext)
    {
      ThreadHelper threadHelper = (ThreadHelper)threadHelperContext;
			try
			{
				MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Load ETF Holdings, Thread {0} started for {1}...", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol));
				LoadETFHoldingsEx(threadHelper.Symbol);
				MDTrace.WriteLine(LogLevel.DEBUG,String.Format("Load ETF Holdings, Thread {0} ended for {1}", Thread.CurrentThread.ManagedThreadId, threadHelper.Symbol));
			}
			finally
			{
			  threadHelper.ResetEvent.Set();
			}
    }
    public static void LoadETFHoldingsEx(String symbol)
    {
        MDTrace.WriteLine(LogLevel.DEBUG,"Retrieving holdings for " + symbol);
        ETFHoldings etfHoldings = MarketDataHelper.GetETFHoldings(symbol);
        if (null == etfHoldings || 0 == etfHoldings.Count)
        {
          MDTrace.WriteLine(LogLevel.DEBUG,"Error retreiving ETF holdings for '"+symbol+"'");
          return;
        }
        if (!ETFHoldingsDA.InsertOrUpdate(etfHoldings))
        {
          MDTrace.WriteLine(LogLevel.DEBUG,"Error adding ETF holdings for " + symbol);
          return;
        }
        MDTrace.WriteLine(LogLevel.DEBUG,"********* E T F  H O L D I N G S  ********"); 
        foreach (ETFHolding etfHolding in etfHoldings)
        {
          MDTrace.WriteLine(LogLevel.DEBUG,etfHolding.ETFSymbol + ","+etfHolding.HoldingSymbol + "," + etfHolding.HoldingSymbolShareClass + "," + etfHolding.HoldingCompanyName);
        }
        MDTrace.WriteLine(LogLevel.DEBUG,"************************");    
    }
  }
}