Skip to content

Fixes DataProcessing #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AlexCatarino merged 1 commit into from
Dec 23, 2025
Merged

Fixes DataProcessing #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 105 additions & 0 deletions DataProcessing/BrainDataConverter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
{
/// <summary>
/// Base implementation for Brain dataset converters.
/// </summary>
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);
}

/// <summary>
/// Converts all available deployment dates.
/// </summary>
public virtual bool ProcessHistory() => throw new NotImplementedException();

/// <summary>
/// Processes a single deployment date file.
/// </summary>
public virtual bool ProcessDate(DateTime date) => throw new NotImplementedException();

public void SaveContentToFile(string ticker, IEnumerable<string> 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<string>(contents);
if (finalFileExists)
{
foreach (var line in File.ReadAllLines(filePath))
{
lines.Add(line);
}
}

File.WriteAllLines(finalPath, [.. lines.OrderBy(x => x[0..8])]);
}

/// <summary>
/// Disposes the S3 client.
/// </summary>
public void Dispose() => S3Client?.DisposeSafely();
}
}
142 changes: 60 additions & 82 deletions DataProcessing/BrainLanguageMetricsEarningsCallsConverter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
{
/// <summary>
/// Converts Brain Language Metrics on Earnings Calls (BLMECT)
Expand All @@ -21,76 +34,57 @@ namespace QuantConnect.DataSource
/// BLMECT/metrics_earnings_call_YYYYMMDD.csv
/// BLMECT/differences_earnings_call_YYYYMMDD.csv
/// </summary>
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
);
}

/// <summary>
/// Converts all available deployment dates.
/// </summary>
public bool ProcessHistory()
public override bool ProcessHistory()
{
var dates = new List<DateTime>();
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);
}

/// <summary>
/// Converts all files for the given deployment date.
/// </summary>
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<string, string[]>();

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;

Expand All @@ -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;
}

Expand All @@ -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<string, List<string>> rowsBySymbol = new();
Dictionary<string, List<string>> 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;
}

Expand All @@ -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;
}

Expand All @@ -170,41 +164,29 @@ public bool ProcessDate(DateTime date)

if (!rowsBySymbol.TryGetValue(ticker, out var list))
{
list = new List<string>();
rowsBySymbol[ticker] = list;
rowsBySymbol[ticker] = list = [];
}

list.Add(outRow);
}
}
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;
}
Expand All @@ -214,15 +196,16 @@ public bool ProcessDate(DateTime date)
/// </summary>
private static string BuildOutputRow(string fileDate, string[] metrics, string[] diff)
{
var f = new List<string>();

// 0 - snapshot date
f.Add(fileDate);
var f = new List<string>
{
// 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]);
Expand All @@ -248,10 +231,5 @@ private static string BuildOutputRow(string fileDate, string[] metrics, string[]

return string.Join(",", f);
}

public void Dispose()
{
// nothing to clean up
}
}
}
Loading