1
0
Fork 0
mirror of https://github.com/maybe-finance/maybe.git synced 2025-07-25 08:09:38 +02:00

Clarify backend data pipeline naming concepts (importers, processors, materializers, calculators, and syncers) (#2255)
Some checks are pending
Publish Docker image / ci (push) Waiting to run
Publish Docker image / Build docker image (push) Blocked by required conditions

* Rename MarketDataSyncer to MarketDataImporter

* Materializers

* Importers

* More reference replacements
This commit is contained in:
Zach Gollwitzer 2025-05-17 16:37:16 -04:00 committed by GitHub
parent b8903d0980
commit 10f255a9a9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 165 additions and 163 deletions

View file

@ -7,7 +7,7 @@
# Each individual account sync will still fetch any missing market data that isn't yet synced, but by running
# this job daily, we significantly reduce overlapping account syncs that both need the same market data (e.g. common security like `AAPL`)
#
class SyncMarketDataJob < ApplicationJob
class ImportMarketDataJob < ApplicationJob
queue_as :scheduled
def perform(opts)
@ -15,6 +15,6 @@ class SyncMarketDataJob < ApplicationJob
mode = opts.fetch(:mode, :full)
clear_cache = opts.fetch(:clear_cache, false)
MarketDataSyncer.new(mode: mode, clear_cache: clear_cache).sync
MarketDataImporter.new(mode: mode, clear_cache: clear_cache).import_all
end
end

View file

@ -0,0 +1,82 @@
class Account::MarketDataImporter
attr_reader :account
def initialize(account)
@account = account
end
def import_all
import_exchange_rates
import_security_prices
end
def import_exchange_rates
return unless needs_exchange_rates?
return unless ExchangeRate.provider
pair_dates = {}
# 1. ENTRY-BASED PAIRS currencies that differ from the account currency
account.entries
.where.not(currency: account.currency)
.group(:currency)
.minimum(:date)
.each do |source_currency, date|
key = [ source_currency, account.currency ]
pair_dates[key] = [ pair_dates[key], date ].compact.min
end
# 2. ACCOUNT-BASED PAIR convert the account currency to the family currency (if different)
if foreign_account?
key = [ account.currency, account.family.currency ]
pair_dates[key] = [ pair_dates[key], account.start_date ].compact.min
end
pair_dates.each do |(source, target), start_date|
ExchangeRate.import_provider_rates(
from: source,
to: target,
start_date: start_date,
end_date: Date.current
)
end
end
def import_security_prices
return unless Security.provider
account_securities = account.trades.map(&:security).uniq
return if account_securities.empty?
account_securities.each do |security|
security.import_provider_prices(
start_date: first_required_price_date(security),
end_date: Date.current
)
security.import_provider_details
end
end
private
# Calculates the first date we require a price for the given security scoped to this account
def first_required_price_date(security)
account.trades.with_entry
.where(security: security)
.where(entries: { account_id: account.id })
.minimum("entries.date")
end
def needs_exchange_rates?
has_multi_currency_entries? || foreign_account?
end
def has_multi_currency_entries?
account.entries.where.not(currency: account.currency).exists?
end
def foreign_account?
account.currency != account.family.currency
end
end

View file

@ -1,82 +0,0 @@
class Account::MarketDataSyncer
attr_reader :account
def initialize(account)
@account = account
end
def sync_market_data
sync_exchange_rates
sync_security_prices
end
private
def sync_exchange_rates
return unless needs_exchange_rates?
return unless ExchangeRate.provider
pair_dates = {}
# 1. ENTRY-BASED PAIRS currencies that differ from the account currency
account.entries
.where.not(currency: account.currency)
.group(:currency)
.minimum(:date)
.each do |source_currency, date|
key = [ source_currency, account.currency ]
pair_dates[key] = [ pair_dates[key], date ].compact.min
end
# 2. ACCOUNT-BASED PAIR convert the account currency to the family currency (if different)
if foreign_account?
key = [ account.currency, account.family.currency ]
pair_dates[key] = [ pair_dates[key], account.start_date ].compact.min
end
pair_dates.each do |(source, target), start_date|
ExchangeRate.sync_provider_rates(
from: source,
to: target,
start_date: start_date,
end_date: Date.current
)
end
end
def sync_security_prices
return unless Security.provider
account_securities = account.trades.map(&:security).uniq
return if account_securities.empty?
account_securities.each do |security|
security.sync_provider_prices(
start_date: first_required_price_date(security),
end_date: Date.current
)
security.sync_provider_details
end
end
# Calculates the first date we require a price for the given security scoped to this account
def first_required_price_date(security)
account.trades.with_entry
.where(security: security)
.where(entries: { account_id: account.id })
.minimum("entries.date")
end
def needs_exchange_rates?
has_multi_currency_entries? || foreign_account?
end
def has_multi_currency_entries?
account.entries.where.not(currency: account.currency).exists?
end
def foreign_account?
account.currency != account.family.currency
end
end

View file

@ -7,8 +7,8 @@ class Account::Syncer
def perform_sync(sync)
Rails.logger.info("Processing balances (#{account.linked? ? 'reverse' : 'forward'})")
sync_market_data
sync_balances
import_market_data
materialize_balances
end
def perform_post_sync
@ -16,9 +16,9 @@ class Account::Syncer
end
private
def sync_balances
def materialize_balances
strategy = account.linked? ? :reverse : :forward
Balance::Syncer.new(account, strategy: strategy).sync_balances
Balance::Materializer.new(account, strategy: strategy).materialize_balances
end
# Syncs all the exchange rates + security prices this account needs to display historical chart data
@ -28,8 +28,8 @@ class Account::Syncer
#
# We rescue errors here because if this operation fails, we don't want to fail the entire sync since
# we have reasonable fallbacks for missing market data.
def sync_market_data
Account::MarketDataSyncer.new(account).sync_market_data
def import_market_data
Account::MarketDataImporter.new(account).import_all
rescue => e
Rails.logger.error("Error syncing market data for account #{account.id}: #{e.message}")
Sentry.capture_exception(e)

View file

@ -1,4 +1,4 @@
class Balance::Syncer
class Balance::Materializer
attr_reader :account, :strategy
def initialize(account, strategy:)
@ -6,9 +6,9 @@ class Balance::Syncer
@strategy = strategy
end
def sync_balances
def materialize_balances
Balance.transaction do
sync_holdings
materialize_holdings
calculate_balances
Rails.logger.info("Persisting #{@balances.size} balances")
@ -23,8 +23,8 @@ class Balance::Syncer
end
private
def sync_holdings
@holdings = Holding::Syncer.new(account, strategy: strategy).sync_holdings
def materialize_holdings
@holdings = Holding::Materializer.new(account, strategy: strategy).materialize_holdings
end
def update_account_info

View file

@ -1,4 +1,4 @@
class ExchangeRate::Syncer
class ExchangeRate::Importer
MissingExchangeRateError = Class.new(StandardError)
MissingStartRateError = Class.new(StandardError)
@ -12,7 +12,7 @@ class ExchangeRate::Syncer
end
# Constructs a daily series of rates for the given currency pair for date range
def sync_provider_rates
def import_provider_rates
if !clear_cache && all_rates_exist?
Rails.logger.info("No new rates to sync for #{from} to #{to} between #{start_date} and #{end_date}, skipping")
return

View file

@ -28,20 +28,20 @@ module ExchangeRate::Provided
end
# @return [Integer] The number of exchange rates synced
def sync_provider_rates(from:, to:, start_date:, end_date:, clear_cache: false)
def import_provider_rates(from:, to:, start_date:, end_date:, clear_cache: false)
unless provider.present?
Rails.logger.warn("No provider configured for ExchangeRate.sync_provider_rates")
Rails.logger.warn("No provider configured for ExchangeRate.import_provider_rates")
return 0
end
ExchangeRate::Syncer.new(
ExchangeRate::Importer.new(
exchange_rate_provider: provider,
from: from,
to: to,
start_date: start_date,
end_date: end_date,
clear_cache: clear_cache
).sync_provider_rates
).import_provider_rates
end
end
end

View file

@ -1,10 +1,12 @@
class Holding::Syncer
# "Materializes" holdings (similar to a DB materialized view, but done at the app level)
# into a series of records we can easily query and join with other data.
class Holding::Materializer
def initialize(account, strategy:)
@account = account
@strategy = strategy
end
def sync_holdings
def materialize_holdings
calculate_holdings
Rails.logger.info("Persisting #{@holdings.size} holdings")

View file

@ -1,4 +1,4 @@
class MarketDataSyncer
class MarketDataImporter
# By default, our graphs show 1M as the view, so by fetching 31 days,
# we ensure we can always show an accurate default graph
SNAPSHOT_DAYS = 31
@ -10,32 +10,32 @@ class MarketDataSyncer
@clear_cache = clear_cache
end
def sync
sync_prices
sync_exchange_rates
def import_all
import_security_prices
import_exchange_rates
end
# Syncs historical security prices (and details)
def sync_prices
def import_security_prices
unless Security.provider
Rails.logger.warn("No provider configured for MarketDataSyncer.sync_prices, skipping sync")
Rails.logger.warn("No provider configured for MarketDataImporter.import_security_prices, skipping sync")
return
end
Security.where.not(exchange_operating_mic: nil).find_each do |security|
security.sync_provider_prices(
security.import_provider_prices(
start_date: get_first_required_price_date(security),
end_date: end_date,
clear_cache: clear_cache
)
security.sync_provider_details(clear_cache: clear_cache)
security.import_provider_details(clear_cache: clear_cache)
end
end
def sync_exchange_rates
def import_exchange_rates
unless ExchangeRate.provider
Rails.logger.warn("No provider configured for MarketDataSyncer.sync_exchange_rates, skipping sync")
Rails.logger.warn("No provider configured for MarketDataImporter.import_exchange_rates, skipping sync")
return
end
@ -43,7 +43,7 @@ class MarketDataSyncer
# pair is a Hash with keys :source, :target, and :start_date
start_date = snapshot? ? default_start_date : pair[:start_date]
ExchangeRate.sync_provider_rates(
ExchangeRate.import_provider_rates(
from: pair[:source],
to: pair[:target],
start_date: start_date,
@ -124,7 +124,7 @@ class MarketDataSyncer
valid_modes = [ :full, :snapshot ]
unless valid_modes.include?(mode.to_sym)
raise InvalidModeError, "Invalid mode for MarketDataSyncer, can only be :full or :snapshot, but was #{mode}"
raise InvalidModeError, "Invalid mode for MarketDataImporter, can only be :full or :snapshot, but was #{mode}"
end
mode.to_sym

View file

@ -1,4 +1,4 @@
class Security::Price::Syncer
class Security::Price::Importer
MissingSecurityPriceError = Class.new(StandardError)
MissingStartPriceError = Class.new(StandardError)
@ -12,7 +12,7 @@ class Security::Price::Syncer
# Constructs a daily series of prices for a single security over the date range.
# Returns the number of rows upserted.
def sync_provider_prices
def import_provider_prices
if !clear_cache && all_prices_exist?
Rails.logger.info("No new prices to sync for #{security.ticker} between #{start_date} and #{end_date}, skipping")
return 0

View file

@ -49,9 +49,9 @@ module Security::Provided
price
end
def sync_provider_details(clear_cache: false)
def import_provider_details(clear_cache: false)
unless provider.present?
Rails.logger.warn("No provider configured for Security.sync_provider_details")
Rails.logger.warn("No provider configured for Security.import_provider_details")
return
end
@ -76,19 +76,19 @@ module Security::Provided
end
end
def sync_provider_prices(start_date:, end_date:, clear_cache: false)
def import_provider_prices(start_date:, end_date:, clear_cache: false)
unless provider.present?
Rails.logger.warn("No provider configured for Security.sync_provider_prices")
Rails.logger.warn("No provider configured for Security.import_provider_prices")
return 0
end
Security::Price::Syncer.new(
Security::Price::Importer.new(
security: self,
security_provider: provider,
start_date: start_date,
end_date: end_date,
clear_cache: clear_cache
).sync_provider_prices
).import_provider_prices
end
private