-Initial pipeline setup for regalytics articles.

Author: caeser1996

.gitignore

@@ -276,4 +276,5 @@ QuantConnect.Lean.sln.DotSettings*
 Research/Notebooks
 
 #Docker result files
-Results/
+Results/
+/venv/

Demonstration.cs

@@ -13,7 +13,7 @@
  * limitations under the License.
  *
 */
-
+/*
 using QuantConnect.Data;
 using QuantConnect.Util;
 using QuantConnect.Orders;
@@ -75,3 +75,4 @@ public override void OnOrderEvent(OrderEvent orderEvent)
         }
     }
 }
+*/

QuantConnect.DataSource.csproj

@@ -3,8 +3,8 @@
   <PropertyGroup>
     <TargetFramework>net5.0</TargetFramework>
     <RootNamespace>QuantConnect.DataSource</RootNamespace>
+    <AssemblyName>QuantConnect.DataSource.RegalyticsArticles</AssemblyName>
   </PropertyGroup>
-
   <ItemGroup>
     <PackageReference Include="QuantConnect.Common" Version="2.5.11800" />
     <PackageReference Include="protobuf-net" Version="3.0.29" />

MyCustomDataType.cs RegalyticsRegulatoryArticle.csMyCustomDataType.cs renamed to RegalyticsRegulatoryArticle.cs

@@ -20,20 +20,69 @@
 using System.IO;
 using QuantConnect.Data;
 using System.Collections.Generic;
+using Newtonsoft.Json;
 
 namespace QuantConnect.DataSource
 {
     /// <summary>
-    /// Example custom data type
+    /// Regalytics Regulatory articles
     /// </summary>
     [ProtoContract(SkipConstructor = true)]
-    public class MyCustomDataType : BaseData
+    public class RegalyticsRegulatoryArticle : BaseData
     {
-        /// <summary>
-        /// Some custom data property
-        /// </summary>
-        [ProtoMember(2000)]
-        public string SomeCustomProperty { get; set; }
+        [JsonProperty(PropertyName = "id")]
+        public int Id { get; set; }
+
+        [JsonProperty(PropertyName = "title")]
+        public string Title { get; set; }
+
+        [JsonProperty(PropertyName = "summary")]
+        public string Summary { get; set; }
+
+        [JsonProperty(PropertyName = "status")]
+        public string Status { get; set; }
+
+        [JsonProperty(PropertyName = "classification")]
+        public string Classificaiton { get; set; }
+
+        [JsonProperty(PropertyName = "filing_type")]
+        public string FilingType { get; set; }
+
+        [JsonProperty(PropertyName = "in_federal_register")]
+        public bool InFederalRegister { get; set; }
+
+        [JsonProperty(PropertyName = "federal_register_number")]
+        public string FederalRegisterNumber { get; set; }
+
+        //[JsonProperty(PropertyName = "regalytics_alert_id")]
+        //public string AlertId { get; set; }
+
+        [JsonProperty(PropertyName = "proposed_comments_due_date")]
+        public DateTime? ProposedCommentsDueDate { get; set; }
+
+        [JsonProperty(PropertyName = "original_publication_date")]
+        public DateTime? OriginalPublicationDate { get; set; }
+
+        [JsonProperty(PropertyName = "federal_register_publication_date")]
+        public DateTime? FederalRegisterPublicationDate { get; set; }
+
+        [JsonProperty(PropertyName = "rule_effective_date")]
+        public DateTime? RuleEffectiveDate { get; set; }
+
+        [JsonProperty(PropertyName = "latest_update")]
+        public DateTime LatestUpdate { get; set; }
+
+        [JsonProperty(PropertyName = "alert_type")]
+        public string AlertType { get; set; }
+
+        [JsonProperty(PropertyName = "states")]
+        public Dictionary<string, List<string>> States { get; set; }
+
+        [JsonProperty(PropertyName = "agencies")]
+        public List<string> Agencies { get; set; }
+
+        [JsonProperty(PropertyName = "pdf_url")]
+        public string AnnouncementUrl { get; set; }
 
         /// <summary>
         /// Return the URL string source of the file. This will be converted to a stream
@@ -48,8 +97,9 @@ public override SubscriptionDataSource GetSource(SubscriptionDataConfig config,
                 Path.Combine(
                     Globals.DataFolder,
                     "alternative",
-                    "mycustomdatatype",
-                    $"{config.Symbol.Value.ToLowerInvariant()}.csv"
+                    "regalytics",
+                    "articles",
+                    $"{date:yyyyMMdd}.json"
                 ),
                 SubscriptionTransportMedium.LocalFile
             );
@@ -65,16 +115,16 @@ public override SubscriptionDataSource GetSource(SubscriptionDataConfig config,
         /// <returns>New instance</returns>
         public override BaseData Reader(SubscriptionDataConfig config, string line, DateTime date, bool isLiveMode)
         {
-            var csv = line.Split(',');
+            var article = JsonConvert.DeserializeObject<RegalyticsRegulatoryArticle>(line);
 
-            var parsedDate = Parse.DateTimeExact(csv[0], "yyyyMMdd");
-            return new MyCustomDataType
-            {
-                Symbol = config.Symbol,
-                SomeCustomProperty = csv[1],
-                Time = parsedDate,
-                EndTime = parsedDate + TimeSpan.FromDays(1)
-            };
+            // date == the day that the data was published (2021-05-21)
+            // 2021-05-21 for example, contains aggregated data from 2021-05-19, 2021-05-20. 
+            // Regalytics publishes at 07:30:00 Eastern time, EndTime should be at that time.
+
+            article.Symbol = config.Symbol;
+            article.EndTime = date.Date.AddHours(7).AddMinutes(30);
+
+            return article;
         }
 
         /// <summary>
@@ -83,12 +133,30 @@ public override BaseData Reader(SubscriptionDataConfig config, string line, Date
         /// <returns>A clone of the object</returns>
         public override BaseData Clone()
         {
-            return new MyCustomDataType
+            return new RegalyticsRegulatoryArticle
             {
                 Symbol = Symbol,
                 Time = Time,
                 EndTime = EndTime,
-                SomeCustomProperty = SomeCustomProperty,
+
+                Id = Id,
+                Title = Title,
+                Summary = Summary,
+                Status = Status,
+                Classificaiton = Classificaiton,
+                FilingType = FilingType,
+                InFederalRegister = InFederalRegister,
+                FederalRegisterNumber = FederalRegisterNumber,
+                // AlertId = AlertId,
+                ProposedCommentsDueDate = ProposedCommentsDueDate,
+                OriginalPublicationDate = OriginalPublicationDate,
+                FederalRegisterPublicationDate = FederalRegisterPublicationDate,
+                RuleEffectiveDate = RuleEffectiveDate,
+                LatestUpdate = LatestUpdate,
+                AlertType = AlertType,
+                States = States,
+                Agencies = Agencies,
+                AnnouncementUrl = AnnouncementUrl,
             };
         }
 
@@ -98,7 +166,7 @@ public override BaseData Clone()
         /// <returns>false</returns>
         public override bool RequiresMapping()
         {
-            return true;
+            return false;
         }
 
         /// <summary>
@@ -116,7 +184,7 @@ public override bool IsSparseData()
         /// </summary>
         public override string ToString()
         {
-            return $"{Symbol} - {SomeCustomProperty}";
+            return $"ID: {Id} - Title: {Title} - Summary: {Summary}";
         }
 
         /// <summary>
@@ -141,7 +209,7 @@ public override List<Resolution> SupportedResolutions()
         /// <returns>The <see cref="T:NodaTime.DateTimeZone" /> of this data type</returns>
         public override DateTimeZone DataTimeZone()
         {
-            return DateTimeZone.Utc;
+            return TimeZones.NewYork;
         }
     }
-}
+}

output/alternative/mycustomdatatype/spy.csv

@@ -0,0 +1,84 @@
+import json
+import pathlib
+from datetime import datetime, timedelta
+import os
+import requests
+
+url = os.environ["REGALYTICS_API_BASE_URL"] + "/all"
+
+payload = json.dumps({
+    "apikey": os.environ["REGALYTICS_API_KEY"]
+})
+headers = {
+    'Content-Type': 'application/json'
+}
+
+response = requests.post(url, headers=headers, data=payload).json()
+articles_path = pathlib.Path('/temp-output-directory/alternative/regalytics/articles')
+# objectives:# download data from API -> temp folder or in memory. Output processed datat  to /temp-output-directory/alternative/regalytics/articles/yyyyMMdd.json
+articles_path.mkdir(parents=True, exist_ok=True)
+# "states": [
+#         {
+#           "name": "United States",
+#           "country": {
+#             "name": "United States"
+#           }
+#         }
+#       ],
+# if states is more than 0
+# loop into state and get the state name
+# 1. query all data, -> /api/.../all; 2. look at latest_update, add delta of 1/2 days;
+# 3. write data to date of latest_update + delta. This date must be on the date we published the article on Regalytics
+articles = response['articles']
+articles_by_date = {}
+
+for article in articles:
+    article['in_federal_register'] = 'yes' in article['in_federal_register'].lower()
+    # State -> Dictionary<string, List<string>>
+    states = {}
+    if 'states' not in article or article['states'] is None:
+        continue
+
+    for state in article['states']:
+        if 'country' not in state:
+            continue
+
+        country = state['country']
+        if country is None:
+            continue
+
+        if not country['name'] in states:
+            country_states = []
+            states[country['name']] = country_states
+        else:
+            country_states = states[country['name']]
+
+        country_states.append(state['name'])
+
+    agencies = [agency['name'] for agency in article['agencies']]
+    article['states'] = states
+    article['agencies'] = agencies
+    # If friday or weekend, the delta will make the date land on Tuesday
+    date = datetime.strptime(article['latest_update'], '%Y-%m-%d')
+    weekday = date.weekday()
+    delta = (8 - weekday) if weekday >= 3 else 2
+    date = date + timedelta(days=delta)
+    date_key = date.strftime('%Y%m%d')
+
+    if date_key not in articles_by_date:
+        date_articles = []
+        articles_by_date[date_key] = date_articles
+    else:
+        date_articles = articles_by_date[date_key]
+
+    date_articles.append(article)
+
+for date, articles in articles_by_date.items():
+    lines = []
+    for article in articles:
+        lines.append(json.dumps(article, indent=None))
+
+    article_lines = '\n'.join(lines)
+
+    with open(articles_path / f'{date}.json', 'w') as article_file:
+        article_file.write(article_lines)

process.py

@@ -13,7 +13,7 @@
  * limitations under the License.
  *
 */
-
+/*
 using System;
 using ProtoBuf;
 using System.IO;
@@ -96,4 +96,4 @@ private BaseData CreateNewInstance()
             };
         }
     }
-}
+}*/