Stock imports (#1363)

* Initial pass

* Marketstack data provider

* Marketstack data provider

* Refactor a bit

.env.example

@@ -15,6 +15,10 @@ PORT=3000
 # This is used to convert between different currencies in the app. In addition, it fetches US stock prices. We use Synth, which is a Maybe product. You can sign up for a free account at synthfinance.com.
 SYNTH_API_KEY=
 
+# Non-US Stock Pricing API
+# This is used to fetch non-US stock prices. We use Marketstack.com for this and while they offer a free tier, it is quite limited. You'll almost certainly need their Basic plan, which is $9.99 per month.
+MARKETSTACK_API_KEY=
+
 # SMTP Configuration
 # This is only needed if you intend on sending emails from your Maybe instance (such as for password resets or email financial reports).
 # Resend.com is a good option that offers a free tier for sending emails.

.env.local.example

@@ -3,3 +3,6 @@ SELF_HOSTED=false
 
 # Enable Synth market data (careful, this will use your API credits)
 SYNTH_API_KEY=yourapikeyhere
+
+# Enable Marketstack market data (careful, this will use your API credits)
+MARKETSTACK_API_KEY=yourapikeyhere

app/controllers/securities_controller.rb

@@ -0,0 +1,5 @@
+class SecuritiesController < ApplicationController
+  def import
+    SecuritiesImportJob.perform_later(params[:exchange_mic])
+  end
+end

app/helpers/securities_helper.rb

@@ -0,0 +1,2 @@
+module SecuritiesHelper
+end

app/jobs/securities_import_job.rb

@@ -0,0 +1,13 @@
+class SecuritiesImportJob < ApplicationJob
+  queue_as :default
+
+  def perform(country_code = nil)
+    exchanges = StockExchange.in_country(country_code)
+    market_stack_client = Provider::Marketstack.new(ENV["MARKETSTACK_API_KEY"])
+
+    exchanges.each do |exchange|
+      importer = Security::Importer.new(market_stack_client, exchange.mic)
+      importer.import
+    end
+  end
+end

app/models/provider/marketstack.rb

@@ -0,0 +1,119 @@
+class Provider::Marketstack
+  include Retryable
+
+  def initialize(api_key)
+    @api_key = api_key
+  end
+
+  def fetch_security_prices(ticker:, start_date:, end_date:)
+    prices = paginate("#{base_url}/eod", {
+      symbols: ticker,
+      date_from: start_date.to_s,
+      date_to: end_date.to_s
+    }) do |body|
+      body.dig("data").map do |price|
+        {
+          date: price["date"],
+          price: price["close"]&.to_f,
+          currency: "USD"
+        }
+      end
+    end
+
+    SecurityPriceResponse.new(
+      prices: prices,
+      success?: true,
+      raw_response: prices.to_json
+    )
+  rescue StandardError => error
+    SecurityPriceResponse.new(
+      success?: false,
+      error: error,
+      raw_response: error
+    )
+  end
+
+  def fetch_tickers(exchange_mic: nil)
+    url = exchange_mic ? "#{base_url}/tickers?exchange=#{exchange_mic}" : "#{base_url}/tickers"
+    tickers = paginate(url) do |body|
+      body.dig("data").map do |ticker|
+        {
+          name: ticker["name"],
+          symbol: ticker["symbol"],
+          exchange: exchange_mic || ticker.dig("stock_exchange", "mic"),
+          country_code: ticker.dig("stock_exchange", "country_code")
+        }
+      end
+    end
+
+    TickerResponse.new(
+      tickers: tickers,
+      success?: true,
+      raw_response: tickers.to_json
+    )
+  rescue StandardError => error
+    TickerResponse.new(
+      success?: false,
+      error: error,
+      raw_response: error
+    )
+  end
+
+  private
+
+    attr_reader :api_key
+
+    SecurityPriceResponse = Struct.new(:prices, :success?, :error, :raw_response, keyword_init: true)
+    TickerResponse = Struct.new(:tickers, :success?, :error, :raw_response, keyword_init: true)
+
+    def base_url
+      "https://api.marketstack.com/v1"
+    end
+
+    def client
+      @client ||= Faraday.new(url: base_url) do |faraday|
+        faraday.params["access_key"] = api_key
+      end
+    end
+
+    def build_error(response)
+      Provider::Base::ProviderError.new(<<~ERROR)
+        Failed to fetch data from #{self.class}
+          Status: #{response.status}
+          Body: #{response.body.inspect}
+      ERROR
+    end
+
+    def fetch_page(url, page, params = {})
+      client.get(url) do |req|
+        params.each { |k, v| req.params[k.to_s] = v.to_s }
+        req.params["offset"] = (page - 1) * 100 # Marketstack uses offset-based pagination
+        req.params["limit"] = 10000 # Maximum allowed by Marketstack
+      end
+    end
+
+    def paginate(url, params = {})
+      results = []
+      page = 1
+      total_results = Float::INFINITY
+
+      while results.length < total_results
+        response = fetch_page(url, page, params)
+
+        if response.success?
+          body = JSON.parse(response.body)
+          page_results = yield(body)
+          results.concat(page_results)
+
+          total_results = body.dig("pagination", "total")
+          page += 1
+        else
+          raise build_error(response)
+        end
+
+        break if results.length >= total_results
+      end
+
+      results
+    end
+end

app/models/security/importer.rb

@@ -0,0 +1,27 @@
+class Security::Importer
+  def initialize(provider, stock_exchange = nil)
+    @provider = provider
+    @stock_exchange = stock_exchange
+  end
+
+  def import
+    securities = @provider.fetch_tickers(exchange_mic: @stock_exchange)&.tickers
+
+    stock_exchanges = StockExchange.where(mic: securities.map { |s| s[:exchange] }).index_by(&:mic)
+    existing_securities = Security.where(ticker: securities.map { |s| s[:symbol] }, stock_exchange_id: stock_exchanges.values.map(&:id)).pluck(:ticker, :stock_exchange_id).to_set
+
+    securities_to_create = securities.map do |security|
+      stock_exchange_id = stock_exchanges[security[:exchange]]&.id
+      next if existing_securities.include?([ security[:symbol], stock_exchange_id ])
+
+      {
+        name: security[:name],
+        ticker: security[:symbol],
+        stock_exchange_id: stock_exchange_id,
+        country_code: security[:country_code]
+      }
+    end.compact
+
+    Security.insert_all(securities_to_create) unless securities_to_create.empty?
+  end
+end

app/models/stock_exchange.rb

@@ -1,2 +1,3 @@
 class StockExchange < ApplicationRecord
+  scope :in_country, ->(country_code) { where(country_code: country_code) }
 end

db/migrate/20241023195438_add_stock_exchange_reference.rb

@@ -0,0 +1,7 @@
+class AddStockExchangeReference < ActiveRecord::Migration[7.2]
+  def change
+    add_column :securities, :country_code, :string
+    add_reference :securities, :stock_exchange, type: :uuid, foreign_key: true
+    add_index :securities, :country_code
+  end
+end

db/schema.rb

@@ -119,7 +119,7 @@ ActiveRecord::Schema[7.2].define(version: 2024_10_24_142537) do
     t.boolean "is_active", default: true, null: false
     t.date "last_sync_date"
     t.uuid "institution_id"
-    t.virtual "classification", type: :string, as: "\nCASE\n    WHEN ((accountable_type)::text = ANY ((ARRAY['Loan'::character varying, 'CreditCard'::character varying, 'OtherLiability'::character varying])::text[])) THEN 'liability'::text\n    ELSE 'asset'::text\nEND", stored: true
+    t.virtual "classification", type: :string, as: "\nCASE\n    WHEN ((accountable_type)::text = ANY (ARRAY[('Loan'::character varying)::text, ('CreditCard'::character varying)::text, ('OtherLiability'::character varying)::text])) THEN 'liability'::text\n    ELSE 'asset'::text\nEND", stored: true
     t.uuid "import_id"
     t.string "mode"
     t.index ["accountable_id", "accountable_type"], name: "index_accounts_on_accountable_id_and_accountable_type"
@@ -478,6 +478,10 @@ ActiveRecord::Schema[7.2].define(version: 2024_10_24_142537) do
     t.string "name"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.string "country_code"
+    t.uuid "stock_exchange_id"
+    t.index ["country_code"], name: "index_securities_on_country_code"
+    t.index ["stock_exchange_id"], name: "index_securities_on_stock_exchange_id"
   end
 
   create_table "security_prices", id: :uuid, default: -> { "gen_random_uuid()" }, force: :cascade do |t|
@@ -599,6 +603,7 @@ ActiveRecord::Schema[7.2].define(version: 2024_10_24_142537) do
   add_foreign_key "imports", "families"
   add_foreign_key "institutions", "families"
   add_foreign_key "merchants", "families"
+  add_foreign_key "securities", "stock_exchanges"
   add_foreign_key "sessions", "impersonation_sessions", column: "active_impersonator_session_id"
   add_foreign_key "sessions", "users"
   add_foreign_key "taggings", "tags"