feat: add initial schwab task and notebook

invaas/cnn_fear_greed_index.py → invaas/schwab/cnn_fear_greed_index.py

@@ -1,5 +1,3 @@
-#! /usr/bin/env python3
-
 import datetime
 import requests
 import typing
@@ -41,12 +39,10 @@ def __call__(self) -> dict:
         return r.json()
 
 
-def get_current_cnn_fear_greed_index(fetcher: Fetcher = None):
+def get_current_cnn_fear_greed_index():
     """Returns CNN's Fear & Greed Index."""
 
-    if fetcher is None:
-        fetcher = Fetcher()
-
+    fetcher = Fetcher()
     response = fetcher()["fear_and_greed"]
     return FearGreedIndex(
         value=response["score"],
@@ -55,10 +51,8 @@ def get_current_cnn_fear_greed_index(fetcher: Fetcher = None):
     )
 
 
-def get_historical_cnn_fear_greed_index(fetcher: Fetcher = None):
+def get_historical_cnn_fear_greed_index():
     """Returns CNN's Fear & Greed Index."""
 
-    if fetcher is None:
-        fetcher = Fetcher()
-
-    return fetcher()["fear_and_greed_historical"]
+    fetcher = Fetcher()
+    return fetcher()["fear_and_greed_historical"]

invaas/schwab/schwab_task.py

@@ -0,0 +1,38 @@
+# Task class for Schwab trading
+
+import requests
+
+from invaas.task import Task
+from invaas.schwab.cnn_fear_greed_index import get_current_cnn_fear_greed_index
+
+
+class SchwabTask(Task):
+    """
+    Task class to execute ETL processes for loading and preparing data.
+    """
+
+    def __init__(self, env: str = "local"):
+        super().__init__(env=env)
+
+        self.product_ids = ["VUG"]
+        self.logger.info(f"Products to trade: {str(self.product_ids)}")
+
+        self.current_fear_greed_index = int(get_current_cnn_fear_greed_index().value)
+        self.logger.info(f"Current fear greed index: {self.current_fear_greed_index}")
+
+        self.min_buy_amount = 2
+        self.max_buy_amount = 100
+        self.logger.info(f"Min buy amount: ${self.min_buy_amount}")
+        self.logger.info(f"Max buy amount: ${self.max_buy_amount}")
+
+    def __get_available_balance(self, account_name: str):
+        pass
+
+    def __buy_product(self, product_id: str, available_cash: float):
+        pass
+
+    def __sell_product(self, product_id: str, owned_crypto: float):
+        pass
+
+    def create_orders(self):
+        pass

invaas/schwab_notebook.ipynb

@@ -0,0 +1,49 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%pip install /Workspace/Shared/.bundle/prod/invaas/files"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import os\n",
+    "import sys\n",
+    "\n",
+    "from dotenv import load_dotenv, find_dotenv\n",
+    "\n",
+    "load_dotenv(find_dotenv())\n",
+    "app_env = os.getenv(\"APP-ENV\")\n",
+    "app_source = os.getenv(\"APP-SOURCE\")\n",
+    "\n",
+    "sys.path.append((app_source if app_env != \"local\" else os.path.abspath(os.path.join(\"..\"))))\n",
+    "\n",
+    "from invaas.schwab.schwab_task import SchwabTask\n",
+    "\n",
+    "task = SchwabTask(env=app_env)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "task.create_orders()"
+   ]
+  }
+ ],
+ "metadata": {
+  "language_info": {
+   "name": "python"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}