diff --git a/DataProcessing/BrainDataConverter.cs b/DataProcessing/BrainDataConverter.cs new file mode 100644 index 0000000..54073f4 --- /dev/null +++ b/DataProcessing/BrainDataConverter.cs @@ -0,0 +1,105 @@ +/* + * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. + * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +using Amazon; +using Amazon.S3; +using QuantConnect.Util; +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; + +namespace QuantConnect.DataProcessing +{ + /// + /// Base implementation for Brain dataset converters. + /// + public class BrainDataConverter : IDisposable + { + protected AmazonS3Client S3Client { get; } + protected string BucketName { get; } + protected string Prefix { get; } + + private readonly string _processedDataDirectory; + private readonly string _destinationDirectory; + + public BrainDataConverter(string prefix, string outputRoot) + { + BucketName = Environment.GetEnvironmentVariable("BRAIN_S3_BUCKET_NAME"); + var awsAccessKeyId = Environment.GetEnvironmentVariable("AWS_ACCESS_KEY_ID"); + var awsSecretAccessKey = Environment.GetEnvironmentVariable("AWS_SECRET_ACCESS_KEY"); + + if (string.IsNullOrWhiteSpace(BucketName) || + string.IsNullOrWhiteSpace(awsAccessKeyId) || + string.IsNullOrWhiteSpace(awsSecretAccessKey)) + { + throw new ArgumentNullException("The BRAIN_S3_BUCKET_NAME, AWS_ACCESS_KEY_ID, or AWS_SECRET_ACCESS_KEY environment variables is not set."); + } + + S3Client = new AmazonS3Client( + awsAccessKeyId, + awsSecretAccessKey, + RegionEndpoint.USEast1 + ); + + prefix = prefix.Trim().ToLowerInvariant(); + Prefix = prefix.ToUpperInvariant(); + + _processedDataDirectory = Path.Combine(Globals.DataFolder, "alternative", "brain", prefix); + _destinationDirectory = Path.Combine(outputRoot, prefix); + Directory.CreateDirectory(_destinationDirectory); + } + + /// + /// Converts all available deployment dates. + /// + public virtual bool ProcessHistory() => throw new NotImplementedException(); + + /// + /// Processes a single deployment date file. + /// + public virtual bool ProcessDate(DateTime date) => throw new NotImplementedException(); + + public void SaveContentToFile(string ticker, IEnumerable contents) + { + ticker = ticker.ToLowerInvariant(); + var filePath = Path.Combine(_processedDataDirectory, $"{ticker}.csv"); + var finalPath = Path.Combine(_destinationDirectory, $"{ticker}.csv"); + + var finalFileExists = File.Exists(filePath); + if (!finalFileExists) + { + filePath = finalPath; + finalFileExists = File.Exists(filePath); + } + + var lines = new HashSet(contents); + if (finalFileExists) + { + foreach (var line in File.ReadAllLines(filePath)) + { + lines.Add(line); + } + } + + File.WriteAllLines(finalPath, [.. lines.OrderBy(x => x[0..8])]); + } + + /// + /// Disposes the S3 client. + /// + public void Dispose() => S3Client?.DisposeSafely(); + } +} \ No newline at end of file diff --git a/DataProcessing/BrainLanguageMetricsEarningsCallsConverter.cs b/DataProcessing/BrainLanguageMetricsEarningsCallsConverter.cs index c0c157c..0ae0f29 100644 --- a/DataProcessing/BrainLanguageMetricsEarningsCallsConverter.cs +++ b/DataProcessing/BrainLanguageMetricsEarningsCallsConverter.cs @@ -1,15 +1,28 @@ +/* + * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. + * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +using Amazon.S3.Model; +using QuantConnect.Logging; +using QuantConnect.Util; using System; using System.Collections.Generic; using System.Globalization; using System.IO; -using Amazon; -using Amazon.S3; -using Amazon.S3.Model; -using QuantConnect.Logging; -using QuantConnect.DataProcessing; using System.Linq; -namespace QuantConnect.DataSource +namespace QuantConnect.DataProcessing { /// /// Converts Brain Language Metrics on Earnings Calls (BLMECT) @@ -21,76 +34,57 @@ namespace QuantConnect.DataSource /// BLMECT/metrics_earnings_call_YYYYMMDD.csv /// BLMECT/differences_earnings_call_YYYYMMDD.csv /// - public class BrainLanguageMetricsEarningsCallsConverter : IBrainDataConverter + public class BrainLanguageMetricsEarningsCallsConverter(string outputRoot) + : BrainDataConverter("blmect", outputRoot) { - private readonly IAmazonS3 _s3; - private readonly string _bucket; - private readonly string _outputRoot; - - public BrainLanguageMetricsEarningsCallsConverter( - string awsAccessKeyId, - string awsSecretAccessKey, - string bucket, - string outputRoot) - { - _bucket = bucket; - _outputRoot = outputRoot; - - _s3 = new AmazonS3Client( - awsAccessKeyId, - awsSecretAccessKey, - RegionEndpoint.USEast1 - ); - } - /// /// Converts all available deployment dates. /// - public bool ProcessHistory() + public override bool ProcessHistory() { var dates = new List(); var req = new ListObjectsV2Request { - BucketName = _bucket, - Prefix = "BLMECT/" + BucketName = BucketName, + Prefix = $"{Prefix}/" }; ListObjectsV2Response resp; do - { - resp = _s3.ListObjectsV2Async(req).GetAwaiter().GetResult(); - var s3Objects = resp.S3Objects; - dates.AddRange(s3Objects.Where(x => x.Key.StartsWith("BLMECT/differences_earnings_call_")).Select(x => DateTime.ParseExact(x.Key[33..41], "yyyyMMdd", CultureInfo.InvariantCulture))); - dates.AddRange(s3Objects.Where(x => x.Key.StartsWith("BLMECT/metrics_earnings_call_")).Select(x => DateTime.ParseExact(x.Key[29..37], "yyyyMMdd", CultureInfo.InvariantCulture))); - req.ContinuationToken = resp.NextContinuationToken; - } + { + resp = S3Client.ListObjectsV2Async(req).GetAwaiter().GetResult(); + var s3Objects = resp.S3Objects; + dates.AddRange(s3Objects.Where(x => x.Key.StartsWith($"{Prefix}/differences_earnings_call_")).Select(x => DateTime.ParseExact(x.Key[33..41], "yyyyMMdd", CultureInfo.InvariantCulture))); + dates.AddRange(s3Objects.Where(x => x.Key.StartsWith($"{Prefix}/metrics_earnings_call_")).Select(x => DateTime.ParseExact(x.Key[29..37], "yyyyMMdd", CultureInfo.InvariantCulture))); + req.ContinuationToken = resp.NextContinuationToken; + } while (resp.IsTruncated); - Log.Trace($"[BLMECT] Found {dates.Distinct().Count()} unique deployment dates."); + Log.Trace($"[{Prefix}] Found {dates.Distinct().Count()} unique deployment dates."); return dates.Distinct().OrderBy(x => x).All(ProcessDate); } /// /// Converts all files for the given deployment date. /// - public bool ProcessDate(DateTime date) + public override bool ProcessDate(DateTime date) { var fileDate = date.ToString("yyyyMMdd", CultureInfo.InvariantCulture); // ------------------------------------------------------------ // 1. Load the DIFF file // ------------------------------------------------------------ - var diffKey = $"BLMECT/differences_earnings_call_{fileDate}.csv"; + var diffKey = $"{Prefix}/differences_earnings_call_{fileDate}.csv"; var diffByTicker = new Dictionary(); - Log.Trace($"[BLMECT] Downloading DIFF: s3://{_bucket}/{diffKey}"); + Log.Trace($"[{Prefix}] Downloading DIFF: s3://***/{diffKey}"); try { - using var diffResponse = _s3.GetObjectAsync(new GetObjectRequest + using var diffResponse = S3Client.GetObjectAsync(new GetObjectRequest { - BucketName = _bucket, + BucketName = BucketName, Key = diffKey }).Result; @@ -105,7 +99,7 @@ public bool ProcessDate(DateTime date) var parts = line.Split(delimiter); if (parts.Length < 47) { - Log.Trace($"[BLMECT] DIFF skipped row (too few columns): {parts.Length}"); + Log.Trace($"[{Prefix}] DIFF skipped row (too few columns): {parts.Length}"); continue; } @@ -116,30 +110,30 @@ public bool ProcessDate(DateTime date) } catch (Exception err) { - Log.Trace($"[BLMECT] DIFF optional file missing for {fileDate}: {err.Message}"); + Log.Trace($"[{Prefix}] DIFF optional file missing for {fileDate}: {err.Message}"); } // ------------------------------------------------------------ // 2. Load the METRICS file // ------------------------------------------------------------ - var metricsKey = $"BLMECT/metrics_earnings_call_{fileDate}.csv"; + var metricsKey = $"{Prefix}/metrics_earnings_call_{fileDate}.csv"; - Log.Trace($"[BLMECT] Downloading METRICS: s3://{_bucket}/{metricsKey}"); + Log.Trace($"[{Prefix}] Downloading METRICS: s3://***/{metricsKey}"); - Dictionary> rowsBySymbol = new(); + Dictionary> rowsBySymbol = []; GetObjectResponse metricsResponse; try { - metricsResponse = _s3.GetObjectAsync(new GetObjectRequest + metricsResponse = S3Client.GetObjectAsync(new GetObjectRequest { - BucketName = _bucket, + BucketName = BucketName, Key = metricsKey }).Result; } catch (Exception err) { - Log.Error(err, $"[BLMECT] Failed to download METRICS file for {fileDate}"); + Log.Error(err, $"[{Prefix}] Failed to download METRICS file for {fileDate}"); return false; } @@ -156,7 +150,7 @@ public bool ProcessDate(DateTime date) var parts = line.Split(delimiter); if (parts.Length < 29) { - Log.Trace($"[BLMECT] METRICS skipped row (too few columns): {parts.Length}"); + Log.Trace($"[{Prefix}] METRICS skipped row (too few columns): {parts.Length}"); continue; } @@ -170,8 +164,7 @@ public bool ProcessDate(DateTime date) if (!rowsBySymbol.TryGetValue(ticker, out var list)) { - list = new List(); - rowsBySymbol[ticker] = list; + rowsBySymbol[ticker] = list = []; } list.Add(outRow); @@ -179,32 +172,21 @@ public bool ProcessDate(DateTime date) } catch (Exception err) { - Log.Error(err, "[BLMECT] Failed parsing METRICS CSV."); + Log.Error(err, $"[{Prefix}] Failed parsing METRICS CSV."); return false; } - var outDir = Path.Combine(_outputRoot, "blmect"); - Directory.CreateDirectory(outDir); - try { - foreach (var kvp in rowsBySymbol) - { - var ticker = kvp.Key.ToLowerInvariant(); - var filePath = Path.Combine(outDir, $"{ticker}.csv"); - - using var writer = new StreamWriter(filePath, append: true); - foreach (var row in kvp.Value) - writer.WriteLine(row); - } + rowsBySymbol.DoForEach(kvp => SaveContentToFile(kvp.Key, kvp.Value)); } catch (Exception err) { - Log.Error(err, "[BLMECT] Failed writing output files."); + Log.Error(err, $"[{Prefix}] Failed writing output files."); return false; } - Log.Trace($"[BLMECT] Completed fileDate={fileDate}: {rowsBySymbol.Count} symbols written."); + Log.Trace($"[{Prefix}] Completed fileDate={fileDate}: {rowsBySymbol.Count} symbols written."); return true; } @@ -214,15 +196,16 @@ public bool ProcessDate(DateTime date) /// private static string BuildOutputRow(string fileDate, string[] metrics, string[] diff) { - var f = new List(); - - // 0 - snapshot date - f.Add(fileDate); + var f = new List + { + // 0 - snapshot date + fileDate, - // metrics fields - f.Add(DateTime.TryParse(metrics[3], out var dt) ? dt.ToString("yyyyMMdd") : ""); - f.Add(metrics[4]); // quarter - f.Add(metrics[5]); // year + // metrics fields + DateTime.TryParse(metrics[3], out var dt) ? dt.ToString("yyyyMMdd") : "", + metrics[4], // quarter + metrics[5] // year + }; for (int i = 6; i < 15; i++) f.Add(metrics[i]); for (int i = 15; i < 20; i++) f.Add(metrics[i]); @@ -248,10 +231,5 @@ private static string BuildOutputRow(string fileDate, string[] metrics, string[] return string.Join(",", f); } - - public void Dispose() - { - // nothing to clean up - } } } diff --git a/DataProcessing/BrainWikipediaPageViewsConverter.cs b/DataProcessing/BrainWikipediaPageViewsConverter.cs index 9b81dd9..6a02bd8 100644 --- a/DataProcessing/BrainWikipediaPageViewsConverter.cs +++ b/DataProcessing/BrainWikipediaPageViewsConverter.cs @@ -1,16 +1,28 @@ +/* + * QUANTCONNECT.COM - Democratizing Finance, Empowering Individuals. + * Lean Algorithmic Trading Engine v2.0. Copyright 2014 QuantConnect Corporation. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. +*/ + +using Amazon.S3.Model; +using QuantConnect.Logging; +using QuantConnect.Util; using System; using System.Collections.Generic; using System.Globalization; using System.IO; -using Amazon; -using Amazon.S3; -using Amazon.S3.Model; -using QuantConnect; -using QuantConnect.Logging; -using QuantConnect.DataProcessing; using System.Linq; -namespace QuantConnect.DataSource +namespace QuantConnect.DataProcessing { /// /// Converts Brain Wikipedia Page Views (BWPV) raw S3 files into Lean-format: @@ -19,78 +31,60 @@ namespace QuantConnect.DataSource /// Raw file pattern: /// s3://{bucket}/BWPV/metrics_YYYYMMDD.csv /// - public class BrainWikipediaPageViewsConverter : IBrainDataConverter + public class BrainWikipediaPageViewsConverter(string outputRoot) + : BrainDataConverter("bwpv", outputRoot) { - private readonly IAmazonS3 _s3Client; - private readonly string _bucket; - private readonly string _outputRoot; - - public BrainWikipediaPageViewsConverter( - string awsAccessKeyId, - string awsSecretAccessKey, - string bucket, - string outputRoot) - { - _bucket = bucket; - _outputRoot = outputRoot; - - _s3Client = new AmazonS3Client( - awsAccessKeyId, - awsSecretAccessKey, - RegionEndpoint.USEast1 - ); - } /// /// Converts all available deployment dates. /// - public bool ProcessHistory() + public override bool ProcessHistory() { var dates = new List(); var req = new ListObjectsV2Request { - BucketName = _bucket, - Prefix = "BWPV/" + BucketName = BucketName, + Prefix = $"{Prefix}/" }; ListObjectsV2Response resp; - + do - { - resp = _s3Client.ListObjectsV2Async(req).GetAwaiter().GetResult(); - var s3Objects = resp.S3Objects; - dates.AddRange(s3Objects.Where(x => x.Key.StartsWith("BWPV/metrics_")).Select(x => DateTime.ParseExact(x.Key[13..21], "yyyyMMdd", CultureInfo.InvariantCulture))); - req.ContinuationToken = resp.NextContinuationToken; - } + { + resp = S3Client.ListObjectsV2Async(req).GetAwaiter().GetResult(); + var s3Objects = resp.S3Objects; + dates.AddRange(s3Objects.Where(x => x.Key.StartsWith($"{Prefix}/metrics_")).Select(x => DateTime.ParseExact(x.Key[13..21], "yyyyMMdd", CultureInfo.InvariantCulture))); + req.ContinuationToken = resp.NextContinuationToken; + } while (resp.IsTruncated); - Log.Trace($"[BWPV] Found {dates.Distinct().Count()} unique deployment dates."); + Log.Trace($"[{Prefix}] Found {dates.Distinct().Count()} unique deployment dates."); return dates.Distinct().OrderBy(x => x).All(ProcessDate); } - + /// /// Processes a single deployment date file. /// - public bool ProcessDate(DateTime date) + public override bool ProcessDate(DateTime date) { - var dateString = date.ToString("yyyyMMdd", CultureInfo.InvariantCulture); - var key = $"BWPV/metrics_{dateString}.csv"; + var fileDate = date.ToString("yyyyMMdd", CultureInfo.InvariantCulture); + var key = $"{Prefix}/metrics_{fileDate}.csv"; - Log.Trace($"[BWPV] Downloading s3://{_bucket}/{key}"); + Log.Trace($"[{Prefix}] Downloading s3://***/{key}"); GetObjectResponse response; try { - response = _s3Client.GetObjectAsync(new GetObjectRequest + response = S3Client.GetObjectAsync(new GetObjectRequest { - BucketName = _bucket, + BucketName = BucketName, Key = key }).Result; } catch (Exception err) { - Log.Error(err, $"[BWPV] Failed to download key {key}"); + Log.Error(err, $"[{Prefix}] Failed to download key {key}"); return false; } @@ -104,7 +98,7 @@ public bool ProcessDate(DateTime date) var header = reader.ReadLine(); if (string.IsNullOrWhiteSpace(header)) { - Log.Error("[BWPV] Empty header line."); + Log.Error($"[{Prefix}] Empty header line."); return false; } @@ -143,8 +137,7 @@ public bool ProcessDate(DateTime date) if (!rowsBySymbol.TryGetValue(ticker, out var list)) { - list = new List(); - rowsBySymbol[ticker] = list; + rowsBySymbol[ticker] = list = []; } list.Add(outRow); @@ -152,41 +145,23 @@ public bool ProcessDate(DateTime date) } catch (Exception err) { - Log.Error(err, "[BWPV] Failed while parsing CSV."); + Log.Error(err, $"[{Prefix}] Failed while parsing CSV."); return false; } - - var outDir = Path.Combine(_outputRoot, "bwpv"); - Directory.CreateDirectory(outDir); - - foreach (var kvp in rowsBySymbol) + try { - var ticker = kvp.Key.ToLowerInvariant(); - var filePath = Path.Combine(outDir, $"{ticker}.csv"); - - try - { - using var writer = new StreamWriter(filePath, append: true); - foreach (var row in kvp.Value) - writer.WriteLine(row); - } - catch (Exception err) - { - Log.Error(err, $"[BWPV] Failed writing file {filePath}"); - return false; - } + rowsBySymbol.DoForEach(kvp => SaveContentToFile(kvp.Key, kvp.Value)); + } + catch (Exception err) + { + Log.Error(err, $"[{Prefix}] Failed writing output files."); + return false; } - return true; - } + Log.Trace($"[{Prefix}] Completed fileDate={fileDate}: {rowsBySymbol.Count} symbols written."); - /// - /// Required by the IBrainDataConverter interface. - /// - public void Dispose() - { - // Nothing to dispose right now. + return true; } } } diff --git a/DataProcessing/Program.cs b/DataProcessing/Program.cs index 0445e7f..893d1dd 100644 --- a/DataProcessing/Program.cs +++ b/DataProcessing/Program.cs @@ -17,9 +17,7 @@ using QuantConnect.Logging; using QuantConnect.Util; using System; -using System.Collections.Generic; using System.IO; -using QuantConnect.DataSource; namespace QuantConnect.DataProcessing { @@ -80,52 +78,27 @@ public static void Main(string[] args) } var deploymentDate = Parse.DateTimeExact(deploymentDateValue, "yyyyMMdd"); - var awsAccessKeyId = Environment.GetEnvironmentVariable("AWS_ACCESS_KEY_ID"); - var awsSecretAccessKey = Environment.GetEnvironmentVariable("AWS_SECRET_ACCESS_KEY"); - var bucket = Environment.GetEnvironmentVariable("BRAIN_S3_BUCKET_NAME"); - - if (string.IsNullOrWhiteSpace(awsAccessKeyId) || - string.IsNullOrWhiteSpace(awsSecretAccessKey) || - string.IsNullOrWhiteSpace(bucket)) - { - Log.Error("Missing AWS credentials or BRAIN_S3_BUCKET."); - Environment.Exit(1); - } - + // ------------------------------------------------------------ // Output directory // ------------------------------------------------------------ - var outputRoot = Path.Combine(Config.Get("temp-output-directory", "/temp-output-directory"),"alternative"); + var outputRoot = Path.Combine(Config.Get("temp-output-directory", "/temp-output-directory"), "alternative", "brain"); Directory.CreateDirectory(outputRoot); Log.Trace($"Starting Brain dataset processor for dataset={dataset}, date={deploymentDate:yyyyMMdd}"); - var converters = new List(); + BrainDataConverter converter = null; try { switch (dataset.ToLowerInvariant()) { case "blmect": - converters.Add( - new BrainLanguageMetricsEarningsCallsConverter( - awsAccessKeyId, - awsSecretAccessKey, - bucket, - outputRoot - ) - ); + converter = new BrainLanguageMetricsEarningsCallsConverter(outputRoot); break; case "bwpv": - converters.Add( - new BrainWikipediaPageViewsConverter( - awsAccessKeyId, - awsSecretAccessKey, - bucket, - outputRoot - ) - ); + converter = new BrainWikipediaPageViewsConverter(outputRoot); break; default: @@ -143,34 +116,31 @@ public static void Main(string[] args) var success = true; - foreach (var converter in converters) + try { - try + if (reprocess) { - if (reprocess) - { - if (!converter.ProcessHistory()) - { - Log.Error($"Failed to process history for dataset={dataset}"); - success = false; - } - } - - if (!converter.ProcessDate(deploymentDate)) + if (!converter.ProcessHistory()) { - Log.Error($"Failed to process dataset={dataset}"); + Log.Error($"Failed to process history for dataset={dataset}"); success = false; } } - catch (Exception e) + + if (!converter.ProcessDate(deploymentDate)) { - Log.Error(e, $"Converter for {dataset} crashed unexpectedly"); + Log.Error($"Failed to process dataset={dataset}"); success = false; } - - converter.DisposeSafely(); + } + catch (Exception e) + { + Log.Error(e, $"Converter for {dataset} crashed unexpectedly"); + success = false; } + converter.DisposeSafely(); + Environment.Exit(success ? 0 : 1); } @@ -180,13 +150,4 @@ private static void PrintHelp() Log.Trace(" dotnet process.dll --dataset "); } } - - /// - /// Shared interface for Brain converters. - /// - public interface IBrainDataConverter : IDisposable - { - bool ProcessDate(DateTime date); - bool ProcessHistory(); - } }