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

Refactor the SECFilings pull

Browse Source
This commit is contained in:
Sean
2025-04-02 11:52:11 -04:00
parent fba7825fab
commit 0315575dda
2 changed files with 81 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
MarketData/MarketDataLib/Helper/SECFilingMarketDataHelper.cs
View File

@@ -1,112 +1,71 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using MarketData.MarketDataModel;
using MarketData.MarketDataModel;
using MarketData.DataAccess;
using MarketData.Utils;
namespace MarketData.Helper
{
public class SECFilingMarketDataHelper
public class SECFilingMarketDataHelper : MarketDataHelperBase<String>
{
private static int MaxThreads = 5; // 10 requests per second is what is allowable under SEC.GOV. We'll request 5 symbols per batch. Note:each request may contain subrequests
private static int WAIT_TIME_MS=1000; // wait between requests
private List<String> symbols;
private int currentIndex = 0;
public SECFilingMarketDataHelper()
{
}
public bool UpdateSECFilings(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(ThreadPoolCallbackUpdateSECFiling, threadHelper);
try { Thread.Sleep(WAIT_TIME_MS); } catch(Exception) { ;} // SEC has a traffic limit
}
MDTrace.WriteLine(LogLevel.DEBUG,"Load SEC Filings, waiting for queued items to complete.");
WaitHandle.WaitAll(resetEvents);
}
return true;
}
catch(Exception exception)
{
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[UpdateSECFilings]Exception {0}",exception.ToString()));
return false;
}
finally
{
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[UpdateSECFilings]End, total took {0}(ms)",profiler.End()));
}
}
public bool UpdateSECFilings()
{
Profiler profiler=new Profiler();
try
{
List<String> watchListSymbols = WatchListDA.GetWatchList("Valuations"); // get the current watchlist symbols
List<String> secFilingSymbols=SECFilingDA.GetDistinctFilingSymbols(); // get the current SEC filing symbols
symbols=watchListSymbols.Union(secFilingSymbols).Distinct().ToList(); // use the union of the two
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(ThreadPoolCallbackUpdateSECFiling, threadHelper);
try { Thread.Sleep(WAIT_TIME_MS); }catch(Exception) { ;} // SEC has a traffic limit
}
MDTrace.WriteLine(LogLevel.DEBUG,"Load SEC Filings, waiting for queued items to complete.");
WaitHandle.WaitAll(resetEvents);
}
return true;
}
catch(Exception exception)
{
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[UpdateSECFilings]Exception {0}",exception.ToString()));
return false;
}
finally
{
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[UpdateSECFilings]End, total took {0}(ms)",profiler.End()));
}
}
private List<String> GetQueueSymbols()
public bool UpdateSECFilings(List<String> symbols)
{
List<String> queueSymbols = new List<String>();
int index = currentIndex;
for (; index < currentIndex + MaxThreads && index < symbols.Count; index++)
Profiler profiler=new Profiler();
try
{
queueSymbols.Add(symbols[index]);
if(null==symbols || 0==symbols.Count)return false;
Queue=symbols;
Index=-1;
ManualResetEvent[] resetEvents = new ManualResetEvent[MaxThreads];
for (int eventIndex = 0; eventIndex < resetEvents.Length; eventIndex++)resetEvents[eventIndex] = new ManualResetEvent(true);
MDTrace.WriteLine(String.Format("Queuing SEC Filings Load ..."));
while (true)
{
ManualResetEvent[] availableEvents=GetAvailableEvents(resetEvents);
ManualResetEvent[] busyEvents=GetBusyEvents(resetEvents);
if (null == PeekQueueItem() && 0==busyEvents.Length)
{
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("SECFilings queue contains {0} items, busy events {1}, all done.",0,busyEvents.Length));
break;
}
for (int index = 0; index < availableEvents.Length; index++)
{
String symbol=GetQueueItem();
if (null != symbol)
{
availableEvents[index].Reset();
ThreadHelper threadHelper = new ThreadHelper(symbol,availableEvents[index]);
ThreadPool.QueueUserWorkItem(ThreadPoolCallbackUpdateSECFiling, threadHelper);
}
else
{
busyEvents=GetBusyEvents(resetEvents);
if(busyEvents.Length!=availableEvents.Length)
{
ManualResetEvent[] resizedEvents=new ManualResetEvent[busyEvents.Length];
Array.Copy(busyEvents, resizedEvents,busyEvents.Length);
resetEvents = resizedEvents;
}
break;
}
} // for
MDTrace.WriteLine(LogLevel.DEBUG,"SECFilings waiting for free slots...");
if(resetEvents.Length>0)WaitHandle.WaitAny(resetEvents);
if(null==PeekQueueItem())resetEvents=ResizeEvents(resetEvents);
} // while
MDTrace.WriteLine(LogLevel.DEBUG,"SECFilings completed.");
return true;
}
finally
{
MDTrace.WriteLine(LogLevel.DEBUG,String.Format("[UpdateSECFilings] End, total took {0}(ms)",profiler.End()));
}
currentIndex = index;
return queueSymbols;
}
public void ThreadPoolCallbackUpdateSECFiling(Object threadHelperContext)
{
ThreadHelper threadHelper = (ThreadHelper)threadHelperContext;
@@ -121,6 +80,7 @@ namespace MarketData.Helper
threadHelper.ResetEvent.Set();
}
}
public static void UpdateSECFiling(String symbol)
{
try
@@ -136,16 +96,16 @@ namespace MarketData.Helper
SECFilings secFilings = MarketDataHelper.GetSECFilings(symbol, cik);
if (null != secFilings)
{
for (int index = 0; index < secFilings.Count; index++)
{
StringBuilder sb = new StringBuilder();
SECFiling secFiling = secFilings[index];
sb.Append(secFiling.Symbol).Append(",");
sb.Append(Utility.DateTimeToStringYYYYHMMHDD(secFiling.FilingDate)).Append(",");
sb.Append(secFiling.SECAccessionNumber).Append(",");
sb.Append(secFiling.Sequence);
MDTrace.WriteLine(LogLevel.DEBUG,sb.ToString());
}
// for (int index = 0; index < secFilings.Count; index++)
// {
// StringBuilder sb = new StringBuilder();
// SECFiling secFiling = secFilings[index];
// sb.Append(secFiling.Symbol).Append(",");
// sb.Append(Utility.DateTimeToStringYYYYHMMHDD(secFiling.FilingDate)).Append(",");
// sb.Append(secFiling.SECAccessionNumber).Append(",");
// sb.Append(secFiling.Sequence);
// MDTrace.WriteLine(LogLevel.DEBUG,sb.ToString());
// }
MDTrace.WriteLine(LogLevel.DEBUG,"Got "+secFilings.Count+" filings for symbol '"+symbol+"'");
SECFilingDA.InsertOrUpdateSECFilings(secFilings);
}