use crate::asset::AssetSymbol;
use crate::config::simple::SimpleConfig;
use crate::config::Config;
use crate::db;
use crate::error::GreedError;
use crate::platform;
use crate::platform::FinancialPlatform;
use crate::strategy::factory::StrategyProviderFactory;
use crate::strategy::provider::StrategyRunnerProvider;
use crate::strategy::runner::{StrategyRunner, TacticStrategyRunner};
use duckdb::Connection;
use log::warn;
use std::ffi::OsStr;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::time::sleep;
#[derive(Debug, Default, PartialEq)]
pub struct GreedRunnerArgs {
pub config_path: PathBuf,
pub is_simulated: bool,
}
pub struct GreedRunner {
run_interval: u64,
config_strategy: Box<dyn StrategyRunner>,
strategy_providers: Vec<Box<dyn StrategyRunnerProvider>>,
config_assets: Vec<AssetSymbol>,
platform: Arc<dyn FinancialPlatform>,
connection: Arc<Mutex<Connection>>,
}
impl GreedRunner {
pub async fn new(
config: Config,
config_path: PathBuf,
platform: Arc<dyn FinancialPlatform>,
connection: Connection,
) -> Result<Self, GreedError> {
let connection = Arc::new(Mutex::new(connection));
let config_strategy = Box::new(TacticStrategyRunner::from_config(
&config,
&platform,
&connection,
));
let factory = StrategyProviderFactory::new(&config, config_path, &platform, &connection);
let strategy_providers = factory.create_providers().await?;
let mut config_assets = Vec::new();
for provider in &strategy_providers {
config_assets.extend(provider.config_assets());
}
config_assets.extend(config.tactics.iter().flat_map(|t| t.assets()));
Ok(Self {
run_interval: config.interval,
config_strategy,
strategy_providers,
config_assets,
platform,
connection,
})
}
pub async fn from_args(
args: GreedRunnerArgs,
connection: Connection,
) -> Result<GreedRunner, GreedError> {
let config_path = args.config_path.clone();
let config = GreedRunner::read_config(&args).await?;
let platform = platform::for_type(&config.platform, args.into())?;
Self::new(config, config_path, platform, connection).await
}
async fn read_config(args: &GreedRunnerArgs) -> Result<Config, GreedError> {
let ext = args.config_path.extension();
if Some(OsStr::new("csv")) == ext {
let simple_config = SimpleConfig::from_path(&args.config_path).await?;
Ok(simple_config.into())
} else {
Config::from_path(&args.config_path).await
}
}
pub async fn run_loop(&self) {
let loop_interval = Duration::from_secs(self.run_interval);
loop {
self.run_tick(loop_interval).await;
sleep(loop_interval).await;
}
}
async fn run_tick(&self, loop_interval: Duration) {
let strategy_runners = match self.provide_strategy_runners().await {
Ok(runners) => runners,
Err(e) => {
warn!("error creating strategy runners: {}", e);
return;
}
};
for runner in strategy_runners {
let _ = runner.run(&self.config_assets).await;
sleep(loop_interval).await;
}
self.config_strategy.run(&self.config_assets).await;
self.record_portfolio_status().await;
}
async fn provide_strategy_runners(&self) -> Result<Vec<Box<dyn StrategyRunner>>, GreedError> {
let mut runners = Vec::new();
for provider in &self.strategy_providers {
let runner = provider.provide_strategy_runner().await?;
runners.push(runner);
}
Ok(runners)
}
async fn record_portfolio_status(&self) {
match self.platform.account().await {
Ok(account) => {
let connection = self
.connection
.lock()
.expect("portfolio status connection lock");
if let Err(e) = db::portfolio_status::insert(&connection, &account) {
warn!("failed to write portfolio status: {}", e);
}
}
Err(e) => warn!("failed to fetch account for portfolio status: {}", e),
}
}
}
#[cfg(test)]
mod test {
use crate::asset::AssetSymbol;
use crate::config::Config;
use crate::db::migrations;
use crate::fixture;
use crate::platform::account::Account;
use crate::platform::quote::Quote;
use crate::platform::MockPlatform;
use crate::run::GreedRunner;
use duckdb::Connection;
use std::path::PathBuf;
use std::time::Duration;
#[tokio::test]
async fn run_tick_writes_portfolio_status_row_per_tick() {
let connection = Connection::open_in_memory().expect("in-memory connection");
migrations::apply(&connection).expect("migrations should apply");
let assert_connection = connection.try_clone().expect("clone connection");
let platform = MockPlatform::new().with_account(Account::fixture()).arc();
let config = Config {
interval: 1,
..Default::default()
};
let runner = GreedRunner::new(config, PathBuf::from("unused.toml"), platform, connection)
.await
.expect("runner should build");
for _ in 0..10 {
runner.run_tick(Duration::from_secs(1)).await;
}
assert_eq!(table_row_count(&assert_connection, "portfolio_status"), 10);
}
#[tokio::test]
async fn run_tick_writes_stock_quote_row_per_tick() {
let connection = Connection::open_in_memory().expect("in-memory connection");
migrations::apply(&connection).expect("migrations should apply");
let assert_connection = connection.try_clone().expect("clone connection");
let mut config = fixture::config("config_single_tactic.toml").await;
config.interval = 0;
let platform = MockPlatform::new()
.with_account(Account::fixture())
.with_quotes(vec![Quote::fixture(AssetSymbol::new("VTI"))])
.arc();
let runner = GreedRunner::new(config, PathBuf::from("unused.toml"), platform, connection)
.await
.expect("runner should build");
for _ in 0..10 {
runner.run_tick(Duration::from_secs(0)).await;
}
assert_eq!(table_row_count(&assert_connection, "portfolio_status"), 10);
assert_eq!(table_row_count(&assert_connection, "stock_quotes"), 10);
}
fn table_row_count(connection: &Connection, table: &str) -> i64 {
connection
.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| {
row.get(0)
})
.expect("table should be queryable")
}
}