diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..a5008a4 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,19 @@ +{ + "cSpell.words": [ + "addstrategy", + "ALTCOINS", + "arange", + "backtrader", + "Bollinger", + "Cardano", + "cerebro", + "Doji", + "fgcior", + "fgis", + "Keltner", + "lovelaces", + "talipp", + "tgens" + ], + "pylint.args": ["--max-line-length=120", "--disable=missing-module-docstring"] +} diff --git a/Dockerfile b/Dockerfile index 429094f..ac17d2c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,6 +14,25 @@ WORKDIR /app # Install dependencies RUN apt-get update && apt-get install -y \ dos2unix \ + wget \ + unzip \ + curl \ + gnupg \ + --no-install-recommends + +# Install Chrome +RUN curl -fsSL https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add - && \ + echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" > /etc/apt/sources.list.d/google-chrome.list && \ + apt-get update && \ + apt-get install -y google-chrome-stable && \ + rm -rf /var/lib/apt/lists/* + +# Download ChromeDriver +RUN CHROME_DRIVER_VERSION=`curl -sS chromedriver.storage.googleapis.com/LATEST_RELEASE` && \ + wget -O /tmp/chromedriver.zip "https://chromedriver.storage.googleapis.com/${CHROME_DRIVER_VERSION}/chromedriver_linux64.zip" && \ + unzip /tmp/chromedriver.zip -d /usr/local/bin/ && \ + rm /tmp/chromedriver.zip + --no-install-recommends # Create a non-privileged user that the app will run under. @@ -22,9 +41,7 @@ ARG UID=10001 RUN adduser \ --disabled-password \ --gecos "" \ - --home "/nonexistent" \ --shell "/sbin/nologin" \ - --no-create-home \ --uid "${UID}" \ appuser @@ -38,7 +55,7 @@ RUN --mount=type=cache,target=/root/.cache/pip \ # Copy the source code into the container. COPY *.py . -COPY strategies/* strategies/ +COPY src/ src/ COPY bot-api.yaml . COPY requirements.txt . COPY .flaskenv . @@ -47,7 +64,6 @@ COPY .flaskenv . COPY *.sh . RUN dos2unix *.sh RUN chmod +x *.sh - RUN /bin/bash -c /app/generate_client.sh # Expose the port that the application listens on. diff --git a/Makefile b/Makefile index 31cbd16..e5859e0 100644 --- a/Makefile +++ b/Makefile @@ -22,6 +22,14 @@ start-b: docker compose up -d --build strategy_b docker compose logs -f strategy_b +start-c: + docker compose up -d --build strategy_c + docker compose logs -f strategy_c + +start-fgis: + docker compose up -d --build fear_and_greed_index_strategy + docker compose logs -f fear_and_greed_index_strategy + start-bb: docker compose up -d --build bollinger_bands_strategy docker compose logs -f bollinger_bands_strategy diff --git a/app.py b/app.py index 49a317f..21d01c9 100644 --- a/app.py +++ b/app.py @@ -1,22 +1,22 @@ -from client import AuthenticatedClient -from client.models import ErrorResponse, Settings -from client.api.settings import get_settings -from client.types import Response -from typing import cast, Union -from flask import Flask, jsonify -import threading -import time +import importlib +import logging import os import sys -import importlib -import yaml +import threading import time -from api import Api -from api import ApiException from datetime import datetime -import logging +from typing import Union, cast + +import yaml +from client import AuthenticatedClient +from client.api.settings import get_settings +from client.models import ErrorResponse, Settings +from client.types import Response +from flask import Flask, jsonify from flask_wtf.csrf import CSRFProtect +from api import Api, ApiException + # Spin up Flask Application app = Flask(__name__) @@ -48,7 +48,7 @@ def health_check(): return jsonify(status='healthy', message='Service is up and running!') def load_strategy(strategy_class): - module = importlib.import_module(f".{strategy_class}", ".strategies") + module = importlib.import_module(f".{strategy_class}", ".src.strategies") if hasattr(module, 'init'): module.init() strategy_class_ref = getattr(module, strategy_class) diff --git a/compose.yaml b/compose.yaml index 255f86d..8c5f5da 100644 --- a/compose.yaml +++ b/compose.yaml @@ -14,7 +14,7 @@ services: coreProvider: maestroToken: <> turboSubmit: false - networkId: "mainnet" # supported: mainnet ot preprod + networkId: "mainnet" # supported: mainnet or preprod logging: - type: {tag: stderr} severity: "Debug" # Options: Debug, Info, Warning or Error @@ -33,7 +33,7 @@ services: environment: BACKEND_URL: http://server:8082 SERVER_API_KEY: ${SERVER_API_KEY} - EXECUTION_DELAY: 90 # Time period in seconds to wait between strategy exeuctions + EXECUTION_DELAY: 90 # Time period in seconds to wait between strategy executions STARTUP_DELAY: 1 # Time period in seconds to wait for the backend to start RETRY_DELAY: 20 # Time period in seconds to wait before retrying to reach the backend CONFIRMATION_DELAY: 90 @@ -49,7 +49,7 @@ services: environment: BACKEND_URL: http://server:8082 SERVER_API_KEY: ${SERVER_API_KEY} - EXECUTION_DELAY: 15 # Time period in seconds to wait between strategy exeuctions + EXECUTION_DELAY: 15 # Time period in seconds to wait between strategy executions STARTUP_DELAY: 1 # Time period in seconds to wait for the backend to start RETRY_DELAY: 20 # Time period in seconds to wait before retrying to reach the backend CONFIRMATION_DELAY: 90 @@ -59,13 +59,54 @@ services: setting_2: 567 ms depends_on: - server + strategy_c: + build: + context: . + environment: + BACKEND_URL: http://server:8082 + SERVER_API_KEY: ${SERVER_API_KEY} + EXECUTION_DELAY: 30 # Time period in seconds to wait between strategy executions + STARTUP_DELAY: 1 # Time period in seconds to wait for the backend to start + RETRY_DELAY: 20 # Time period in seconds to wait before retrying to reach the backend + CONFIRMATION_DELAY: 90 + STRATEGY: strategy_c + CONFIG: | + ASSET_PAIR: "asset1266q2ewhgul7jh3xqpvjzqarrepfjuler20akr-asset1xdz4yj4ldwlpsz2yjgjtt9evg9uskm8jrzjwhj" + START_TIME: "2023-06-15T19:19:56.462Z" + END_TIME: "2024-06-15T19:19:56.462Z" + BIN_INTERVAL: "1d" + depends_on: + - server + fear_and_greed_index_strategy: + build: + context: . + environment: + BACKEND_URL: http://server:8082 + SERVER_API_KEY: ${SERVER_API_KEY} + EXECUTION_DELAY: 60 # Time period in seconds to wait between strategy executions + STARTUP_DELAY: 1 # Time period in seconds to wait for the backend to start + RETRY_DELAY: 20 # Time period in seconds to wait before retrying to reach the backend + CONFIRMATION_DELAY: 90 + STRATEGY: fear_and_greed_index_strategy + CONFIG: | + BASE_ASSET: lovelace + # GENS for MAINNET: + TARGET_ASSET: dda5fdb1002f7389b33e036b6afee82a8189becb6cba852e8b79b4fb.0014df1047454e53 + # tGENS for PRERPOD: + # TARGET_ASSET: c6e65ba7878b2f8ea0ad39287d3e2fd256dc5c4160fc19bdf4c4d87e.7447454e53 + POSITION_SIZE_LOVELACES: 1000000 + STD_DEV_MULTIPLIER: 1.5 + PERIOD: 5 + FEAR_AND_GREED_INDEX_THRESHOLD: 60 + depends_on: + - server bollinger_bands_strategy: build: context: . environment: BACKEND_URL: http://server:8082 SERVER_API_KEY: ${SERVER_API_KEY} - EXECUTION_DELAY: 20 # Time period in seconds to wait between strategy exeuctions + EXECUTION_DELAY: 20 # Time period in seconds to wait between strategy executions STARTUP_DELAY: 1 # Time period in seconds to wait for the backend to start RETRY_DELAY: 20 # Time period in seconds to wait before retrying to reach the backend CONFIRMATION_DELAY: 90 @@ -104,5 +145,7 @@ services: RSI_OVERSOLD: 35 # Higher threshold for oversold BB_PERIOD: 20 # Standard period BB_STD_DEV: 1.8 # Tighter bands for volatility + USE_FEAR_AND_GREED: true + FEAR_AND_GREED_INDEX_THRESHOLD: 60 depends_on: - server diff --git a/requirements.txt b/requirements.txt index de8bc7a..2d245ab 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,9 @@ +backtrader==1.9.78.123 +beautifulsoup4==4.12.3 flask==3.0 +Flask-WTF==1.2.1 gunicorn==22.0.0 openapi-python-client==0.19.1 +selenium==4.21.0 talipp==2.2.0 -Flask-WTF==1.2.1 +webdriver-manager==4.0.1 diff --git a/strategies/__init__.py b/src/__init__.py similarity index 100% rename from strategies/__init__.py rename to src/__init__.py diff --git a/src/data_extraction/__init__.py b/src/data_extraction/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/data_extraction/fear_and_greed_index_web_scraper.py b/src/data_extraction/fear_and_greed_index_web_scraper.py new file mode 100644 index 0000000..c05bdb4 --- /dev/null +++ b/src/data_extraction/fear_and_greed_index_web_scraper.py @@ -0,0 +1,138 @@ +import logging +import re +from typing import Optional + +from bs4 import BeautifulSoup +from selenium import webdriver +from selenium.webdriver.chrome.service import Service +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait +from webdriver_manager.chrome import ChromeDriverManager + + +class ScraperException(Exception): + """Custom exception for scraper-related errors.""" + + +class FearAndGreedIndexWebScraper: + """ + A scraper for the Cardano Fear and Greed Index. + + This class provides functionality to scrape the Cardano Fear and Greed Index + from a specified URL using Selenium WebDriver and BeautifulSoup. + + Attributes: + URL (str): The URL of the Cardano Fear and Greed Index page. + CLASS_SELECTOR (str): The CSS class selector for the index value element. + WEB_DRIVER_WAIT_DEFAULT_TIMEOUT (int): Default timeout for WebDriverWait. + """ + + URL: str = 'https://cfgi.io/cardano-fear-greed-index/' + # cspell:disable-next-line + CLASS_SELECTOR: str = 'apexcharts-datalabel-value' + WEB_DRIVER_WAIT_DEFAULT_TIMEOUT: int = 15 + + def __init__(self, logger: logging.Logger, web_driver_wait_timeout: Optional[int] = None): + """ + Initialize the FearGreedIndexScraper. + + Args: + logger (logging.Logger): Logger object for logging messages. + web_driver_wait_timeout (Optional[int]): + Timeout for WebDriverWait. Defaults to WEB_DRIVER_WAIT_DEFAULT_TIMEOUT. + """ + self.logger: logging.Logger = logger + self.index_value: Optional[str] = None + self.web_driver_wait_timeout: int = web_driver_wait_timeout or self.WEB_DRIVER_WAIT_DEFAULT_TIMEOUT + self.driver: webdriver.Chrome = self._init_driver() + + def _init_driver(self) -> webdriver.Chrome: + """ + Initialize and return a Chrome WebDriver. + + Returns: + webdriver.Chrome: An instance of Chrome WebDriver. + """ + options = webdriver.ChromeOptions() + options.add_argument('--headless') + options.add_argument('--no-sandbox') + options.add_argument('--disable-dev-shm-usage') + + return webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options) + + def fetch_page_content(self) -> str: + """ + Fetch the page content from the URL. + + Returns: + str: The HTML content of the page. + + Raises: + ScraperException: If there's an error fetching the page content. + """ + try: + self.driver.get(self.URL) + WebDriverWait(self.driver, self.web_driver_wait_timeout).until( + EC.presence_of_element_located((By.CLASS_NAME, self.CLASS_SELECTOR)) + ) + return self.driver.page_source + except Exception as e: + self.logger.error(f"Error fetching page content: {e}") + raise ScraperException(f"Failed to fetch page content: {e}") from e + + def parse_index_value(self, html_content: str) -> None: + """ + Parse the index value from the HTML content. + + Args: + html_content (str): The HTML content to parse. + + Raises: + ScraperException: If there's an error parsing the HTML content or if the index element is not found. + """ + try: + soup = BeautifulSoup(html_content, 'html.parser') + index_element = soup.find(class_=self.CLASS_SELECTOR) + if index_element: + self.index_value = index_element.get_text(strip=True) + self.logger.info(f"Successfully parsed index value: {self.index_value}") + else: + raise ScraperException("Could not find the Fear and Greed Index element on the page.") + except Exception as e: + self.logger.error(f"Error parsing HTML content: {e}") + raise ScraperException(f"Failed to parse HTML content: {e}") from e + + @staticmethod + def extract_number(percentage_str: str) -> Optional[int]: + """ + Extract the numeric value from a percentage string. + + Args: + percentage_str (str): The percentage string to extract the number from. + + Returns: + Optional[int]: The extracted number as an integer, or None if no number is found. + """ + match = re.search(r'\d+', percentage_str) + return int(match.group()) if match else None + + def get_index_value(self) -> Optional[int]: + """ + Get the Fear and Greed Index value. + + Returns: + Optional[int]: The Fear and Greed Index value as an integer, or None if the value couldn't be retrieved. + """ + try: + with self.driver: # Use context manager for proper cleanup + html_content = self.fetch_page_content() + if html_content: + self.parse_index_value(html_content) + return self.extract_number(self.index_value) if self.index_value else None + except ScraperException as e: + self.logger.error(f"Scraper error: {e}") + return None + finally: + if self.driver: + self.driver.quit() diff --git a/src/data_extraction/genius_yield_api_scraper.py b/src/data_extraction/genius_yield_api_scraper.py new file mode 100644 index 0000000..fadb682 --- /dev/null +++ b/src/data_extraction/genius_yield_api_scraper.py @@ -0,0 +1,94 @@ +from datetime import datetime, timedelta +from typing import Dict, List, Optional + +import requests + + +class GeniusYieldAPIScraper: + """ + A class to interact with the GeniusYield API for fetching and parsing kline data. + + Attributes: + base_url (str): The base URL for the GeniusYield API. + asset_pair (str): The asset pair for which to fetch data. + params (Dict[str, str]): Parameters for the API request. + """ + + base_url: str = "https://api.geniusyield.co" + + def __init__(self, asset_pair: str, start_time: str, end_time: str, bin_interval: str): + """ + Initialize the GeniusYieldAPI instance. + + Args: + asset_pair (str): The asset pair for which to fetch data. + start_time (str): The start time for the data range. + end_time (str): The end time for the data range. + bin_interval (str): The interval for data binning. + """ + self.asset_pair: str = asset_pair + self.params: Dict[str, str] = { + 'startTime': start_time, + 'endTime': end_time, + 'binInterval': bin_interval + } + + def fetch_data(self) -> Optional[List[Dict[str, str]]]: + """ + Fetch kline data from the GeniusYield API. + + Returns: + Optional[List[Dict[str, str]]]: A list of kline data points if successful, None otherwise. + + Raises: + requests.exceptions.RequestException: If there's an error during the API request. + """ + url: str = f"{self.base_url}/market/by-asset-pair/{self.asset_pair}/kline" + try: + response: requests.Response = requests.get(url, params=self.params, timeout=15) + response.raise_for_status() # Check if the request was successful + data: Dict[str, List[Dict[str, str]]] = response.json() + return data.get('data') + except requests.exceptions.RequestException as e: + print(f"Error fetching data: {e}") + return None + + def parse_data(self, data: List[Dict[str, str]]) -> List[Dict[str, float]]: + """ + Parse the kline data received from the API. + + Args: + data (List[Dict[str, str]]): The raw kline data from the API. + + Returns: + List[Dict[str, float]]: A list of parsed kline data points. + """ + parsed_data: List[Dict[str, float]] = [] + for item in data: + end_timestamp: str = item.get('time', '') + timestamp_dt: datetime = datetime.strptime(end_timestamp, '%Y-%m-%dT%H:%M:%S.%fZ') + new_timestamp_dt: datetime = timestamp_dt + timedelta(days=1) + new_timestamp_str: str = new_timestamp_dt.strftime('%Y-%m-%dT%H:%M:%S.%fZ') + + kline: Dict[str, float] = { + 'start_timestamp': end_timestamp, + 'end_timestamp': new_timestamp_str, + 'open': float(item.get('open', 0)), + 'high': float(item.get('high', 0)), + 'low': float(item.get('low', 0)), + 'close': float(item.get('close', 0)) + } + parsed_data.append(kline) + return parsed_data + + def get_kline_data(self) -> Optional[List[Dict[str, float]]]: + """ + Fetch and parse kline data from the GeniusYield API. + + Returns: + Optional[List[Dict[str, float]]]: A list of parsed kline data points if successful, None otherwise. + """ + raw_data: Optional[List[Dict[str, str]]] = self.fetch_data() + if raw_data: + return self.parse_data(raw_data) + return None diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/models/candlestick.py b/src/models/candlestick.py new file mode 100644 index 0000000..d701d34 --- /dev/null +++ b/src/models/candlestick.py @@ -0,0 +1,140 @@ +from datetime import datetime +from typing import ClassVar, Dict, Optional + + +class Candlestick: + """ + Represents a single candlestick in a financial chart. + + This class encapsulates the data for a candlestick, including timestamps, + open, high, low, and close prices. + + Attributes: + start_timestamp (str): The start time of the candlestick period. + end_timestamp (str): The end time of the candlestick period. + open (float): The opening price of the candlestick. + high (float): The highest price during the candlestick period. + low (float): The lowest price during the candlestick period. + close (float): The closing price of the candlestick. + """ + + TIMESTAMP_FORMAT: ClassVar[str] = '%Y-%m-%dT%H:%M:%S.%fZ' + REQUIRED_FIELDS: ClassVar[list] = ['start_timestamp', 'end_timestamp', 'open', 'high', 'low', 'close'] + + def __init__(self, candlestick_data: Dict[str, float]): + """ + Initialize a Candlestick instance. + + Args: + candlestick_data (Dict[str, float]): A dictionary containing the candlestick data. + + Raises: + ValueError: If the input data is invalid or missing required fields. + """ + if not isinstance(candlestick_data, dict): + raise ValueError("data must be a dictionary") + + for field in self.REQUIRED_FIELDS: + if field not in candlestick_data: + raise ValueError(f"Missing required field: {field}") + + self.start_timestamp: str = candlestick_data['start_timestamp'] + self.end_timestamp: str = candlestick_data['end_timestamp'] + self.open: float = candlestick_data['open'] + self.high: float = candlestick_data['high'] + self.low: float = candlestick_data['low'] + self.close: float = candlestick_data['close'] + + self._validate_data() + + def _validate_data(self) -> None: + """ + Validate the candlestick data. + + Raises: + ValueError: If any of the data validations fail. + """ + # Validate timestamps + try: + start_dt = datetime.strptime(self.start_timestamp, self.TIMESTAMP_FORMAT) + end_dt = datetime.strptime(self.end_timestamp, self.TIMESTAMP_FORMAT) + except ValueError as exc: + raise ValueError(f"Timestamps must be in the format '{self.TIMESTAMP_FORMAT}'") from exc + + # Ensure start_timestamp is less than end_timestamp + if start_dt >= end_dt: + raise ValueError("start_timestamp must be less than end_timestamp") + + # Validate that other attributes are numbers + for attr in ["open", "high", "low", "close"]: + value = getattr(self, attr) + if not isinstance(value, (int, float)): + raise ValueError(f"{attr} must be a number") + + # Ensure open, close values are within the high and low range + if not (self.low <= self.open <= self.high and self.low <= self.close <= self.high): + raise ValueError("Invalid values for open, high, low, or close") + + @classmethod + def create_if_valid(cls, cpc: Dict[str, float]) -> Optional['Candlestick']: + """ + Create a Candlestick instance if the data is valid. + + Args: + cpc (Dict[str, float]): A dictionary containing the candlestick data. + + Returns: + Optional[Candlestick]: A Candlestick instance if the data is valid, None otherwise. + """ + try: + return cls(cpc) + except ValueError: + return None + + def get_duration(self) -> float: + """ + Calculate the duration of the candlestick period. + + Returns: + float: The duration in seconds. + """ + start_dt = datetime.strptime(self.start_timestamp, self.TIMESTAMP_FORMAT) + end_dt = datetime.strptime(self.end_timestamp, self.TIMESTAMP_FORMAT) + return (end_dt - start_dt).total_seconds() + + def get_midpoint(self) -> float: + """ + Calculate the midpoint price of the candlestick. + + Returns: + float: The midpoint between the high and low prices. + """ + return (self.high + self.low) / 2 + + def is_bullish(self) -> bool: + """ + Determine if the candlestick is bullish (closing price higher than opening price). + + Returns: + bool: True if bullish, False otherwise. + """ + return self.close > self.open + + def get_price_range(self) -> float: + """ + Calculate the price range of the candlestick. + + Returns: + float: The difference between the high and low prices. + """ + return self.high - self.low + + def __repr__(self) -> str: + """ + Return a string representation of the Candlestick. + + Returns: + str: A string representation of the Candlestick instance. + """ + return (f"Candlestick(start={self.start_timestamp}, end={self.end_timestamp}, " + f"open={self.open}, high={self.high}, low={self.low}, close={self.close})") diff --git a/src/models/candlestick_price_chart.py b/src/models/candlestick_price_chart.py new file mode 100644 index 0000000..db5d244 --- /dev/null +++ b/src/models/candlestick_price_chart.py @@ -0,0 +1,106 @@ +from typing import Dict, List, Optional, Tuple + +from src.models.candlestick import Candlestick + + +class CandlestickPriceChart: + """ + A class representing a candlestick price chart. + + This class processes a list of candlestick price data and creates a chart + of valid candlesticks. + + Attributes: + candlesticks (List[Candlestick]): A list of valid Candlestick objects. + """ + + CHART_WIDTH_MULTIPLIER = 2 + CHART_WIDTH_OFFSET = 1 + + def __init__(self, candlestick_price_chart_data: List[Dict[str, float]]): + """ + Initialize the CandlestickPriceChart with the given price chart data. + + Args: + candlestick_price_chart_data (List[Dict[str, float]]): A list of dictionaries + containing candlestick price data. + + Raises: + ValueError: If the input data is empty or invalid. + """ + if not candlestick_price_chart_data: + raise ValueError("Input data cannot be empty") + + self.candlesticks = [Candlestick.create_if_valid(cpc) for cpc in candlestick_price_chart_data] + self.candlesticks = [cs for cs in self.candlesticks if cs is not None] + + if not self.candlesticks: + raise ValueError("No valid candlesticks could be created from the input data") + + @property + def chart_height(self) -> int: + """ + Get the height of the chart. + + Returns: + int: The number of candlesticks in the chart. + """ + return len(self.candlesticks) + + @property + def chart_width(self) -> int: + """ + Get the width of the chart. + + Returns: + int: The width of the chart, calculated as twice the number of + candlesticks plus one. + """ + return len(self.candlesticks) * self.CHART_WIDTH_MULTIPLIER + self.CHART_WIDTH_OFFSET + + def get_price_range(self) -> Tuple[float, float]: + """ + Get the price range of the chart. + + Returns: + Tuple[float, float]: A tuple containing the minimum and maximum prices in the chart. + """ + min_price = min(c.low for c in self.candlesticks) + max_price = max(c.high for c in self.candlesticks) + return min_price, max_price + + def get_candlestick_at_index(self, index: int) -> Optional[Candlestick]: + """ + Get the candlestick at a specific index. + + Args: + index (int): The index of the candlestick to retrieve. + + Returns: + Optional[Candlestick]: The Candlestick object at the given index, or None if the index is out of range. + """ + if 0 <= index < len(self.candlesticks): + return self.candlesticks[index] + return None + + def get_average_price(self) -> float: + """ + Calculate the average price across all candlesticks. + + Returns: + float: The average price. + """ + total_price = sum((c.open + c.close) / 2 for c in self.candlesticks) + return total_price / len(self.candlesticks) + + def is_uptrend(self) -> bool: + """ + Determine if the chart shows an uptrend. + + Returns: + bool: True if the closing price of the last candlestick is higher than the opening price of the first, + False otherwise. + """ + if len(self.candlesticks) < 2: + return False + return self.candlesticks[-1].close > self.candlesticks[0].open diff --git a/src/strategies/__init__.py b/src/strategies/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/strategies/bollinger_bands_strategy.py b/src/strategies/bollinger_bands_strategy.py new file mode 100644 index 0000000..feed6c4 --- /dev/null +++ b/src/strategies/bollinger_bands_strategy.py @@ -0,0 +1,203 @@ +import logging +import time +from datetime import datetime +from typing import Optional, Tuple + +from talipp.indicators import BB + +from api import Api +from src.models.candlestick import Candlestick +from src.utils.logger_utils import LoggerUtils +from src.utils.market_maker import MarketMaker + + +# pylint: disable=invalid-name +class bollinger_bands_strategy: + """ + A trading strategy based on Bollinger Bands. + + This class implements a trading strategy that uses Bollinger Bands to make + buy and sell decisions in a given market. + + Attributes: + api_client (Api): The API client for market interactions. + logger (logging.Logger): Logger for outputting information and errors. + market_maker (MarketMaker): Handles order placement and management. + bb (BB): Bollinger Bands indicator. + initialized (bool): Whether the strategy has been initialized. + last_candle (Optional[Candle]): The last processed candle. + last_execution_time (Optional[datetime]): Timestamp of the last execution. + """ + + def __init__(self, api_client: Api, config: dict, logger: logging.Logger): + """ + Initialize the Bollinger Bands strategy. + + Args: + api_client (Api): The API client for market interactions. + config (dict): Configuration parameters for the strategy. + logger (logging.Logger): Logger for outputting information and errors. + """ + logger.info(" > init: bollinger_bands_strategy instance created.") + LoggerUtils.log_warning(logger) + + self.api_client = api_client + self.logger = logger + self.initialized = False + self.last_candle: Optional[Candlestick] = None + self.last_execution_time: Optional[datetime] = None + self._values: Tuple[Optional[float], Optional[float]] = (None, None) + + # Strategy Configuration + self.position_size = float(config["POSITION_SIZE_LOVELACES"]) + self.std_dev_multiplier = float(config["STD_DEV_MULTIPLIER"]) + self.period = int(config["PERIOD"]) + self.base_asset = config["BASE_ASSET"] + self.target_asset = config["TARGET_ASSET"] + self.market = f"{self.base_asset}_{self.target_asset}" + + self._log_configuration() + + self.bb = BB(self.period, self.std_dev_multiplier) + self.market_maker = MarketMaker(api_client, config, logger) + + def _log_configuration(self) -> None: + """Log the strategy configuration.""" + self.logger.info(" STRATEGY CONFIGURATION:") + self.logger.info(f" > base_asset : {self.base_asset}") + self.logger.info(f" > target_asset : {self.target_asset}") + self.logger.info(f" > market : {self.market}") + self.logger.info(f" > position_size : {self.position_size}") + self.logger.info(f" > std_dev_multiplier : {self.std_dev_multiplier}") + self.logger.info(f" > period : {self.period}") + + def process_candle(self, candle: Candlestick) -> None: + """ + Process a new candle and make trading decisions. + + Args: + candle (Candle): The new candle to process. + """ + if self.initialized: + self.logger.info(f" > processing candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}") + else: + self.logger.info( + f" > processing init candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}" + ) + + if self.last_candle and self.last_candle.timestamp == candle.timestamp: + self.logger.info(" > Candle has already been processed. Nothing to do.") + return + + self.last_candle = candle + + value = float(candle.base_close) + self.bb.add(value) + self._values = (self._values[-1], value) + + if len(self.bb) < 2 or self.bb[-1] is None or self.bb[-2] is None: + self._log_initialization_status() + return + + self._log_bollinger_bands() + + if not self.initialized: + self.logger.info(" -> Initialization phase. Do not place orders yet.") + return + + self._check_and_place_orders(candle) + self.market_maker.log_orders() + + def _log_initialization_status(self) -> None: + """Log the initialization status of Bollinger Bands.""" + self.logger.info(" BOLLINGER BANDS: Initializing... ⚙️ ⏳ ") + self.logger.info(" > Upper band: Not available.") + self.logger.info(" > Lower band: Not available.") + + def _log_bollinger_bands(self) -> None: + """Log the current Bollinger Bands values.""" + self.logger.info(" BOLLINGER BANDS: ") + self.logger.info(f" > Upper band: {self.bb[-1].ub}") + self.logger.info(f" > Lower band: {self.bb[-1].lb}") + + def _check_and_place_orders(self, candle: Candlestick) -> None: + """ + Check Bollinger Bands crossovers and place orders accordingly. + + Args: + candle (Candle): The current candle being processed. + """ + if self._values[-2] >= self.bb[-2].lb and self._values[-1] < self.bb[-1].lb: + self._handle_buy_signal(candle) + elif self._values[-2] <= self.bb[-2].ub and self._values[-1] > self.bb[-1].ub: + self._handle_sell_signal(candle) + + def _handle_buy_signal(self, candle: Candlestick) -> None: + """Handle a buy signal.""" + self.logger.info(" -> Price moved below the lower band -> BUY! 🛒 🛒 🛒 ") + self.market_maker.cancel_sell_orders() + if not self.market_maker.get_buy_orders(): + self.market_maker.place_buy_order(candle.base_close) + else: + self.logger.info(" > Already placed BUY order. Nothing to do.") + + def _handle_sell_signal(self, candle: Candlestick) -> None: + """Handle a sell signal.""" + self.logger.info(" -> Price moved above the upper band -> SELL! 💲 💲 💲 ") + self.market_maker.cancel_buy_orders() + if not self.market_maker.get_sell_orders(): + self.market_maker.place_sell_order(candle.base_close) + else: + self.logger.info(" > Already placed SELL order. Nothing to do.") + + # pylint: disable=unused-argument + def execute(self, api_client: Api, config: dict, logger: logging.Logger) -> None: + """ + Execute the strategy. + + This method is called periodically to process new market data and make trading decisions. + + Args: + api_client (Api): The API client for market interactions. + config (dict): Configuration parameters for the strategy. + logger (logging.Logger): Logger for outputting information and errors. + """ + current_time = datetime.now() + + if self.last_execution_time is None: + self._initialize_strategy(api_client) + else: + time_since_last_execution = (current_time - self.last_execution_time).total_seconds() + logger.info(f"Last executed: {self.last_execution_time}") + logger.info(f"Seconds since last execution: {time_since_last_execution} seconds") + + self.last_execution_time = current_time + self.initialized = True + + try: + get_market_price = api_client.get_market_price(self.market) + print(get_market_price) + candle = get_market_price[0] + self.process_candle(candle) + # pylint: disable=broad-exception-caught + except Exception as e: + logger.error(" > ⚠️ [FAILED] could not process candle ⚠️") + logger.exception(f" > Exception: {str(e)}") + + def _initialize_strategy(self, api_client: Api) -> None: + """ + Initialize the strategy with historical data. + + Args: + api_client (Api): The API client for fetching historical data. + """ + self.logger.info("Executing for the first time -> initialize.") + candles = api_client.get_price_history(self.market, resolution="1m", sort="asc", limit=self.period*5) + for candle in candles[:-1]: + self.logger.info("--------------------------------------------------------------------------------") + self.process_candle(candle) + time.sleep(1) + self.logger.info(" > [OK] Initialized.") + self.logger.info("========================================================================") + self.initialized = True + self.last_candle = None diff --git a/src/strategies/combined_rsi_bollinger_strategy.py b/src/strategies/combined_rsi_bollinger_strategy.py new file mode 100644 index 0000000..cc27609 --- /dev/null +++ b/src/strategies/combined_rsi_bollinger_strategy.py @@ -0,0 +1,308 @@ +import logging +import time +from datetime import datetime +from typing import Dict, Optional + +from talipp.indicators import BB, RSI + +from api import Api +from src.data_extraction.fear_and_greed_index_web_scraper import ( + FearAndGreedIndexWebScraper, +) +from src.models.candlestick import Candlestick +from src.utils.logger_utils import LoggerUtils +from src.utils.market_maker import MarketMaker + + +# pylint: disable=invalid-name +class combined_rsi_bollinger_strategy: + """ + A trading strategy combining RSI, Bollinger Bands, and optionally Fear & Greed Index. + + This class implements a trading strategy that uses RSI and Bollinger Bands indicators, + along with an optional Fear & Greed Index to make buy and sell decisions in a given market. + + Attributes: + api_client (Api): The API client for market interactions. + logger (logging.Logger): Logger for outputting information and errors. + market_maker (MarketMaker): Handles order placement and management. + rsi (RSI): RSI indicator instance. + bb (BB): Bollinger Bands indicator instance. + fgis (Optional[FearGreedIndexScraper]): Fear & Greed Index scraper, if enabled. + """ + + def __init__(self, api_client: Api, config: Dict[str, any], logger: logging.Logger): + """ + Initialize the Combined RSI Bollinger strategy. + + Args: + api_client (Api): The API client for market interactions. + config (Dict[str, any]): Configuration parameters for the strategy. + logger (logging.Logger): Logger for outputting information and errors. + + Raises: + ValueError: If the configuration is invalid. + """ + logger.info(" > init: combined_rsi_bollinger_strategy instance created.") + LoggerUtils.log_warning(logger) + + self.api_client = api_client + self.logger = logger + self.initialized = False + self.last_candle: Optional[Candlestick] = None + self.last_execution_time: Optional[datetime] = None + self.cached_fear_and_greed_index = None + + try: + self.validate_config(config) + except ValueError as e: + logger.error(f"Configuration validation failed: {str(e)}") + raise + + self._initialize_strategy_parameters(config) + self._log_strategy_configuration() + self._initialize_indicators(config) + + def _initialize_strategy_parameters(self, config: Dict[str, any]) -> None: + """Initialize strategy parameters from the configuration.""" + self.position_size = float(config["POSITION_SIZE_LOVELACES"]) + self.rsi_period = int(config["RSI_PERIOD"]) + self.rsi_overbought = float(config["RSI_OVERBOUGHT"]) + self.rsi_oversold = float(config["RSI_OVERSOLD"]) + self.bb_period = int(config["BB_PERIOD"]) + self.bb_std_dev = float(config["BB_STD_DEV"]) + self.base_asset = config["BASE_ASSET"] + self.target_asset = config["TARGET_ASSET"] + self.market = f"{self.base_asset}_{self.target_asset}" + self.use_fear_and_greed = config.get("USE_FEAR_AND_GREED", False) + self.fear_and_greed_index_threshold = int(config["FEAR_AND_GREED_INDEX_THRESHOLD"]) + + def _log_strategy_configuration(self) -> None: + """Log the strategy configuration.""" + self.logger.info(" STRATEGY CONFIGURATION:") + self.logger.info(f" > base_asset : {self.base_asset}") + self.logger.info(f" > target_asset : {self.target_asset}") + self.logger.info(f" > market : {self.market}") + self.logger.info(f" > position_size : {self.position_size}") + self.logger.info(f" > rsi_period : {self.rsi_period}") + self.logger.info(f" > rsi_overbought : {self.rsi_overbought}") + self.logger.info(f" > rsi_oversold : {self.rsi_oversold}") + self.logger.info(f" > bb_period : {self.bb_period}") + self.logger.info(f" > bb_std_dev : {self.bb_std_dev}") + self.logger.info(f" > use_fear_and_greed : {self.use_fear_and_greed}") + if self.use_fear_and_greed: + self.logger.info(f" > fear_greed_threshold: {self.fear_and_greed_index_threshold}") + + def _initialize_indicators(self, config: Dict[str, any]) -> None: + """Initialize strategy indicators and components.""" + self.rsi = RSI(self.rsi_period) + self.bb = BB(self.bb_period, self.bb_std_dev) + self.market_maker = MarketMaker(self.api_client, config, self.logger) + self.fgis = FearAndGreedIndexWebScraper(self.logger) if self.use_fear_and_greed else None + + @staticmethod + def validate_config(config: Dict[str, any]) -> None: + """ + Validate the configuration parameters. + + Args: + config (Dict[str, any]): The configuration dictionary to validate. + + Raises: + ValueError: If any configuration parameter is invalid or missing. + """ + required_fields = [ + "POSITION_SIZE_LOVELACES", "RSI_PERIOD", "RSI_OVERBOUGHT", "RSI_OVERSOLD", + "BB_PERIOD", "BB_STD_DEV", "BASE_ASSET", "TARGET_ASSET", "FEAR_AND_GREED_INDEX_THRESHOLD" + ] + + for field in required_fields: + if field not in config: + raise ValueError(f"Missing required configuration field: {field}") + + # Validate types and ranges + try: + assert float(config["POSITION_SIZE_LOVELACES"]) > 0, "POSITION_SIZE_LOVELACES must be positive" + assert 2 <= int(config["RSI_PERIOD"]) <= 100, "RSI_PERIOD must be between 2 and 100" + assert 50 <= float(config["RSI_OVERBOUGHT"]) <= 100, "RSI_OVERBOUGHT must be between 50 and 100" + assert 0 <= float(config["RSI_OVERSOLD"]) <= 50, "RSI_OVERSOLD must be between 0 and 50" + assert 2 <= int(config["BB_PERIOD"]) <= 100, "BB_PERIOD must be between 2 and 100" + assert 0 < float(config["BB_STD_DEV"]) <= 5, "BB_STD_DEV must be between 0 and 5" + assert 0 <= int(config["FEAR_AND_GREED_INDEX_THRESHOLD"]) <= 100, \ + "FEAR_AND_GREED_INDEX_THRESHOLD must be between 0 and 100" + assert isinstance(config["BASE_ASSET"], str) and config["BASE_ASSET"].strip(), \ + "BASE_ASSET should be a non-empty string" + assert isinstance(config["TARGET_ASSET"], str) and config["TARGET_ASSET"].strip(), \ + "TARGET_ASSET should be a non-empty string" + except ValueError as e: + raise ValueError(f"Invalid configuration value: {str(e)}") from e + except AssertionError as e: + raise ValueError(f"Configuration validation failed: {str(e)}") from e + + def get_fear_and_greed_index(self) -> Optional[int]: + """ + Get the current Fear and Greed Index value. + + Returns: + Optional[int]: The Fear and Greed Index value, or None if not available or not used. + """ + if not self.use_fear_and_greed or self.fgis is None: + return None + return self.fgis.get_index_value() or self.fear_and_greed_index_threshold + + def process_candle(self, candle: Candlestick) -> None: + """ + Process a new candle and make trading decisions. + + Args: + candle (Candle): The new candle to process. + """ + if self.initialized: + self.logger.info(f" > processing candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}") + else: + self.logger.info( + f" > processing init candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}" + ) + + if self.last_candle and self.last_candle.timestamp == candle.timestamp: + self.logger.info(" > Candle has already been processed. Nothing to do.") + return + + self.last_candle = candle + + value = float(candle.base_close) + self.rsi.add(value) + self.bb.add(value) + + if len(self.rsi) < self.rsi_period or len(self.bb) < self.bb_period: + self.logger.info( + f" Indicators: Initializing... RSI({len(self.rsi)}/{self.rsi_period}), \ + BB({len(self.bb)}/{self.bb_period}) ⚙️ ⏳ " + ) + return + + self._log_indicator_values(value) + + if not self.initialized: + self.logger.info(" -> Initialization phase. Do not place orders yet.") + return + + self._execute_trading_logic(value) + + # pylint: disable=unused-argument + def _log_indicator_values(self, value: float) -> None: + """Log the current values of the indicators.""" + current_rsi = self.rsi[-1] + current_bb = self.bb[-1] + middle_band = (current_bb.ub + current_bb.lb) / 2 + fear_and_greed_index = None + if self.initialized: + # Prevent unnecessary calls to external website. + fear_and_greed_index = self.get_fear_and_greed_index() + self.cached_fear_and_greed_index = fear_and_greed_index + + self.logger.info(f" RSI: {current_rsi:.2f}") + self.logger.info(f" BB: Lower {current_bb.lb:.2f}, Middle {middle_band:.2f}, Upper {current_bb.ub:.2f}") + if fear_and_greed_index is not None: + self.logger.info(f" Fear & Greed Index: {fear_and_greed_index}") + else: + self.logger.info(" Fear & Greed Index: Not available") + + def _execute_trading_logic(self, value: float) -> None: + """Execute the trading logic based on indicator values.""" + current_rsi = self.rsi[-1] + current_bb = self.bb[-1] + fear_and_greed_index = self.cached_fear_and_greed_index or self.get_fear_and_greed_index() + + buy_signal = (current_rsi < self.rsi_oversold and value <= current_bb.lb) + sell_signal = (current_rsi > self.rsi_overbought and value >= current_bb.ub) + + if fear_and_greed_index is not None: + buy_signal = buy_signal and (fear_and_greed_index < self.fear_and_greed_index_threshold) + sell_signal = sell_signal and (fear_and_greed_index > 100 - self.fear_and_greed_index_threshold) + + if buy_signal: + self._handle_buy_signal(fear_and_greed_index) + elif sell_signal: + self._handle_sell_signal(fear_and_greed_index) + else: + self.logger.info(" -> No clear signal or conflicting indicators. Holding position.") + + self.market_maker.log_orders() + + def _handle_buy_signal(self, fear_and_greed_index: Optional[int]) -> None: + """Handle a buy signal.""" + self.logger.info(" -> Strong BUY signal: RSI oversold, price below lower BB" + + (f", high fear (index: {fear_and_greed_index})" if fear_and_greed_index is not None else "")) + self.market_maker.cancel_sell_orders() + if not self.market_maker.get_buy_orders(): + self.market_maker.place_buy_order(self.last_candle.base_close) + else: + self.logger.info(" > Already placed BUY order. Nothing to do.") + + def _handle_sell_signal(self, fear_and_greed_index: Optional[int]) -> None: + """Handle a sell signal.""" + self.logger.info(" -> Strong SELL signal: RSI overbought, price above upper BB" + + (f", high greed (index: {fear_and_greed_index})" if fear_and_greed_index is not None else "")) + self.market_maker.cancel_buy_orders() + if not self.market_maker.get_sell_orders(): + self.market_maker.place_sell_order(self.last_candle.base_close) + else: + self.logger.info(" > Already placed SELL order. Nothing to do.") + + # pylint: disable=unused-argument + def execute(self, api_client: Api, config: Dict[str, any], logger: logging.Logger) -> None: + """ + Execute the strategy. + + This method is called periodically to process new market data and make trading decisions. + + Args: + api_client (Api): The API client for market interactions. + config (Dict[str, any]): Configuration parameters for the strategy. + logger (logging.Logger): Logger for outputting information and errors. + """ + current_time = datetime.now() + + if self.last_execution_time is None: + self._initialize_strategy(api_client) + else: + time_since_last_execution = (current_time - self.last_execution_time).total_seconds() + logger.info(f"Last executed: {self.last_execution_time}") + logger.info(f"Seconds since last execution: {time_since_last_execution} seconds") + + self.last_execution_time = current_time + self.initialized = True + + try: + market_price = api_client.get_market_price(self.market) + print(market_price) + candle = market_price[0] + self.process_candle(candle) + # pylint: disable=broad-exception-caught + except Exception as e: + logger.error(" > ⚠️ [FAILED] could not process candle ⚠️") + logger.exception(f" > Exception: {str(e)}") + + def _initialize_strategy(self, api_client: Api) -> None: + """ + Initialize the strategy with historical data. + + Args: + api_client (Api): The API client for fetching historical data. + """ + self.logger.info("Executing for the first time -> initialize.") + candles = api_client.get_price_history( + self.market, + resolution="1m", + sort="asc", + limit=max(self.rsi_period, self.bb_period) * 5 + ) + for candle in candles[:-1]: + self.logger.info("--------------------------------------------------------------------------------") + self.process_candle(candle) + time.sleep(1) + self.logger.info(" > [OK] Initialized.") + self.logger.info("========================================================================") + self.initialized = True + self.last_candle = None diff --git a/src/strategies/fear_and_greed_index_strategy.py b/src/strategies/fear_and_greed_index_strategy.py new file mode 100644 index 0000000..bbe5eec --- /dev/null +++ b/src/strategies/fear_and_greed_index_strategy.py @@ -0,0 +1,146 @@ +import logging +from typing import Dict, Optional + +from api import Api +from src.data_extraction.fear_and_greed_index_web_scraper import ( + FearAndGreedIndexWebScraper, +) +from src.models.candlestick import Candlestick +from src.utils.logger_utils import LoggerUtils +from src.utils.market_maker import MarketMaker + + +# pylint: disable=invalid-name +class fear_and_greed_index_strategy: + """ + A trading strategy based on the Fear and Greed Index. + + This class implements a trading strategy that uses the Fear and Greed Index + to make buy and sell decisions in a given market. + + Attributes: + api_client (Api): The API client for market interactions. + logger (logging.Logger): Logger for outputting information and errors. + market_maker (MarketMaker): Handles order placement and management. + fgis (FearGreedIndexScraper): Scraper for the Fear and Greed Index. + """ + + def __init__(self, api_client: Api, config: Dict[str, any], logger: logging.Logger): + """ + Initialize the Fear and Greed Index strategy. + + Args: + api_client (Api): The API client for market interactions. + config (Dict[str, any]): Configuration parameters for the strategy. + logger (logging.Logger): Logger for outputting information and errors. + """ + logger.info(" > init: fear_and_greed_index_strategy instance created.") + LoggerUtils.log_warning(logger) + + self.api_client = api_client + self.logger = logger + + self._initialize_strategy_parameters(config) + self._log_strategy_configuration() + + self.market_maker = MarketMaker(api_client, config, logger) + self.fgis = FearAndGreedIndexWebScraper(self.logger) + + def _initialize_strategy_parameters(self, config: Dict[str, any]) -> None: + """Initialize strategy parameters from the configuration.""" + self.fear_and_greed_index_threshold = int(config["FEAR_AND_GREED_INDEX_THRESHOLD"]) + self.base_asset = config["BASE_ASSET"] + self.target_asset = config["TARGET_ASSET"] + self.market = f"{self.base_asset}_{self.target_asset}" + + def _log_strategy_configuration(self) -> None: + """Log the strategy configuration.""" + self.logger.info(" STRATEGY CONFIGURATION:") + self.logger.info(f" > base_asset : {self.base_asset}") + self.logger.info(f" > target_asset : {self.target_asset}") + self.logger.info(f" > market : {self.market}") + self.logger.info(f" > fear_greed_threshold: {self.fear_and_greed_index_threshold}") + + # pylint: disable=unused-argument + def execute(self, api_client: Api, config: Dict[str, any], logger: logging.Logger) -> None: + """ + Execute the strategy. + + This method is called periodically to process new market data and make trading decisions. + + Args: + api_client (Api): The API client for market interactions. + config (Dict[str, any]): Configuration parameters for the strategy. + logger (logging.Logger): Logger for outputting information and errors. + """ + try: + self.logger.info("Executing Strategy...") + index_value = self.fgis.get_index_value() + if index_value: + self._process_index_value(index_value) + else: + self.logger.info('Failed to retrieve the Fear and Greed Index.') + # pylint: disable=broad-exception-caught + except Exception as e: + self.logger.error(" > ⚠️ [FAILED] could not process market data ⚠️") + self.logger.exception(f" > Exception: {str(e)}") + + def _process_index_value(self, index_value: int) -> None: + """ + Process the Fear and Greed Index value and make trading decisions. + + Args: + index_value (int): The current Fear and Greed Index value. + """ + market_price = self._get_market_price() + if market_price is None: + return + + if index_value > self.fear_and_greed_index_threshold: + self._handle_greed_signal(market_price) + else: + self._handle_fear_signal(market_price) + + self.market_maker.log_orders() + + def _get_market_price(self) -> Optional[Candlestick]: + """ + Get the current market price. + + Returns: + Optional[Candle]: The current market price as a Candle object, or None if retrieval fails. + """ + try: + return self.api_client.get_market_price(self.market)[0] + # pylint: disable=broad-exception-caught + except Exception as e: + self.logger.error(f"Failed to retrieve market price: {str(e)}") + return None + + def _handle_greed_signal(self, market_price: Candlestick) -> None: + """ + Handle a greed signal (buy altcoins). + + Args: + market_price (Candle): The current market price. + """ + self.logger.info(" -> Greed? -> BUY ALTCOINS ! 🛒 🛒 🛒 ") + self.market_maker.cancel_sell_orders() + if self.market_maker.get_buy_orders(): + self.logger.info(" > Already placed BUY order. Nothing to do.") + else: + self.market_maker.place_buy_order(market_price.base_close) + + def _handle_fear_signal(self, market_price: Candlestick) -> None: + """ + Handle a fear signal (sell altcoins). + + Args: + market_price (Candle): The current market price. + """ + self.logger.info(" -> Fear? -> SELL ALTCOINS! 💲 💲 💲 ") + self.market_maker.cancel_buy_orders() + if self.market_maker.get_sell_orders(): + self.logger.info(" > Already placed SELL order. Nothing to do.") + else: + self.market_maker.place_sell_order(market_price.base_close) diff --git a/strategies/strategy_a.py b/src/strategies/strategy_a.py similarity index 80% rename from strategies/strategy_a.py rename to src/strategies/strategy_a.py index 106c7cc..740d728 100644 --- a/strategies/strategy_a.py +++ b/src/strategies/strategy_a.py @@ -1,19 +1,22 @@ -from datetime import datetime import math +from datetime import datetime + from api import ApiException +from src.utils.logger_utils import LoggerUtils + def render_cardano_asset_name_with_policy(policy_asset_string): if 'lovelace' in policy_asset_string.lower(): return "ADA" - + # Split the input string by the period to separate the policy ID and the asset name parts = policy_asset_string.split('.') if len(parts) != 2: return "Invalid input format. Expected format: ." - + # Extract the asset name part hex_asset_name = parts[1] - + try: # Attempt to decode from hexadecimal to bytes, then decode bytes to a UTF-8 string readable_string = bytes.fromhex(hex_asset_name).decode('utf-8') @@ -23,30 +26,22 @@ def render_cardano_asset_name_with_policy(policy_asset_string): except ValueError: # In case of a ValueError, the input was not valid hexadecimal readable_string = f"Invalid hexadecimal: {hex_asset_name}" - + return readable_string +# pylint: disable=invalid-name class strategy_a: - def __init__(self, api_client, CONFIG, logger): + + # pylint: disable=unused-argument + def __init__(self, api_client, config, logger): self.start_time = datetime.now() self.counter = 0 self.last_execution_time = None self.last_order_ref=None logger.info(" > init: strategy_a instance created.") - - logger.info("========================================================================") - logger.info(" ") - logger.info(" ⚠️ WARNING! ⚠️ ") - logger.info(" ") - logger.info(" THIS IS ONLY A PROOF-OF-CONCEPT EXAMPLE STRATEGY IMPLEMENTATION. ") - logger.info(" ") - logger.info(" IT IS ONLY INTENDED AS IMPLEMENTATION REFERENCE FOR TRADING STRATEGIES.") - logger.info(" ") - logger.info(" THIS IMPLEMENTATION IS NOT PRODUCTION-READY. ") - logger.info(" ") - logger.info("========================================================================") - - + LoggerUtils.log_warning(logger) + + logger.info("==============================================") logger.info(" SETTINGS ") logger.info("==============================================") @@ -67,70 +62,72 @@ def __init__(self, api_client, CONFIG, logger): try: markets = api_client.get_markets() for market in markets: + # pylint: disable=line-too-long logger.info(f" > Market: {render_cardano_asset_name_with_policy(market.base_asset)} / {render_cardano_asset_name_with_policy(market.target_asset)}") logger.debug(f" > {market.market_id}") except ApiException as e: logger.exception(f"ApiException: HTTP {e.status_code}: {e.response}") - + logger.info("==============================================") logger.info(" tGENS ASSET DETAILS ") logger.info("==============================================") tGENS="c6e65ba7878b2f8ea0ad39287d3e2fd256dc5c4160fc19bdf4c4d87e.7447454e53" try: tgens_asset = api_client.get_asset(tGENS) - logger.info(f" Asset Details - tGENS:") + logger.info(" Asset Details - tGENS:") logger.info(f" > Ticker: {tgens_asset.asset_ticker}") logger.info(f" > Decimals: {tgens_asset.asset_decimals}") except ApiException as e: logger.exception(f"ApiException: HTTP {e.status_code}: {e.response}") - + logger.info("==============================================") logger.info(" WALLET BALANCE ") logger.info("==============================================") try: balances = api_client.get_balances() - logger.info(f" Balances:") + logger.info(" Balances:") logger.info(f" > Balances: {balances}") logger.info(f" > ADA balance: {math.floor(int(balances.get('lovelace', 0)) / 1_000_000)} ₳") except ApiException as e: logger.exception(f"ApiException: HTTP {e.status_code}: {e.response}") - + logger.info("==============================================") logger.info(" TRADING FEES ") logger.info("==============================================") try: fees = api_client.get_trading_fees() - logger.info(f" Trading Fees:") + logger.info(" Trading Fees:") logger.info(f" > Maker Fee: {int(fees.flat_maker_fee) / 1_000_000 } ₳ + {fees.percentage_maker_fee}%") logger.info(f" > Taker Fee: {int(fees.flat_taker_fee) / 1_000_000 } ₳ + {fees.percentage_taker_fee}%") except ApiException as e: logger.exception(f"ApiException: HTTP {e.status_code}: {e.response}") - + logger.info("==============================================") gens_ada_market_id="lovelace_c6e65ba7878b2f8ea0ad39287d3e2fd256dc5c4160fc19bdf4c4d87e.7447454e53" try: order_book_response = api_client.get_order_book(gens_ada_market_id) - logger.info(f" ADA/GENS Order Book:") - - logger.info(f" > ASKS:") + logger.info(" ADA/GENS Order Book:") + + logger.info(" > ASKS:") for order in order_book_response.asks: logger.info(f" > ask > Amount: {order.offer_amount}, Price: {order.price}") - logger.info(f" > BIDS:") + logger.info(" > BIDS:") for order in order_book_response.bids: logger.info(f" > bid > Amount: {order.offer_amount}, Price: {order.price}") except ApiException as e: logger.exception(f"ApiException: HTTP {e.status_code}: {e.response}") - + logger.info("==============================================") logger.info(" OWN ORDERS ") logger.info("==============================================") + # pylint: disable=unused-variable market_id="lovelace_c6e65ba7878b2f8ea0ad39287d3e2fd256dc5c4160fc19bdf4c4d87e.7447454e53" - + try: response = api_client.get_own_orders(gens_ada_market_id) - logger.info(f" Own orders:") - + logger.info(" Own orders:") + for order in response.asks: logger.info(f" > ask > Amount: {order.offer_amount}, Price: {order.price}") @@ -138,13 +135,13 @@ def __init__(self, api_client, CONFIG, logger): logger.info(f" > bid > Amount: {order.offer_amount}, Price: {order.price}") except ApiException as e: logger.exception(f"ApiException: HTTP {e.status_code}: {e.response}") - + logger.info("==============================================") logger.info(" GENS PRICE HISTORY ") logger.info("==============================================") tgens_market_id="lovelace_c6e65ba7878b2f8ea0ad39287d3e2fd256dc5c4160fc19bdf4c4d87e.7447454e53" gens_market_id="lovelace_dda5fdb1002f7389b33e036b6afee82a8189becb6cba852e8b79b4fb.0014df1047454e53" - + try: response = api_client.get_price_history(tgens_market_id, "1h", "2024-01-01", "2024-01-02") for candle in response: @@ -152,7 +149,7 @@ def __init__(self, api_client, CONFIG, logger): except ApiException as e: logger.exception(f"ApiException: HTTP {e.status_code}: {e.response}") - def execute(self, api_client, CONFIG, logger): + def execute(self, api_client, config, logger): current_time = datetime.now() if self.last_execution_time is None: @@ -168,35 +165,37 @@ def execute(self, api_client, CONFIG, logger): self.counter += 1 logger.info(f" > Counter: {self.counter}") - if self.last_order_ref == None: + if self.last_order_ref is None: # No order was placed -> place order. - logger.info(f" > PLACING ORDER....") - try: + logger.info(" > PLACING ORDER....") + try: response = api_client.place_order( offered_amount="1", offered_token="lovelace", price_token="c6e65ba7878b2f8ea0ad39287d3e2fd256dc5c4160fc19bdf4c4d87e.7447454e53", price_amount="1" ) - logger.info(f" > [OK] PLACED NEW ORDER") + logger.info(" > [OK] PLACED NEW ORDER") logger.info(f"order_ref: {response.order_ref}") logger.info(f"nft_token: {response.nft_token}") self.last_order_ref = response.order_ref + # pylint: disable=bare-except except: - logger.exception(f" > [FAILED] could not place order. ❌") - + logger.exception(" > [FAILED] could not place order. ❌") + else: # An order has already been placed -> cancel it. ref = self.last_order_ref shortened_ref = f"{ref[:8]}...{ref[-8:]}" if len(ref) > 16 else ref logger.info(f" > CANCELING {shortened_ref}") - try: + try: response = api_client.cancel_order(order_reference=self.last_order_ref) logger.info(f" > [OK] cancelled: {shortened_ref}") + # pylint: disable=bare-except except: - logger.exception(f" > [FAILED] could not cancel: {shortened_ref} ❌") - + logger.exception(" > [FAILED] could not cancel: {shortened_ref} ❌") + # Reset the reference, so we place a new order. self.last_order_ref=None - - logger.info(f" > DONE.") + + logger.info(" > DONE.") diff --git a/src/strategies/strategy_b.py b/src/strategies/strategy_b.py new file mode 100644 index 0000000..1d455a7 --- /dev/null +++ b/src/strategies/strategy_b.py @@ -0,0 +1,35 @@ +from datetime import datetime + +from src.utils.logger_utils import LoggerUtils + + +# pylint: disable=invalid-name +class strategy_b: + def __init__(self, client, config, logger): + self.first_execution_time = None + self.client = client + self.config = config + self.logger = logger + logger.info("Strategy B instance created") + LoggerUtils.log_warning(logger) + + # pylint: disable=unused-argument + def execute(self, client, config, logger): + current_time = datetime.now() + if self.first_execution_time is None: + self.first_execution_time = current_time + + # Calculate the time difference in seconds + time_since_first_execution = (current_time - self.first_execution_time).total_seconds() + + # Toggle behavior every 30 seconds + if int(time_since_first_execution // 30) % 2 == 0: + self.behavior_a() + else: + self.behavior_b() + + def behavior_a(self): + self.logger.info("Executing Strategy B: Behavior A 🍎") + + def behavior_b(self): + self.logger.info("Executing Strategy B: Behavior B 🍌") diff --git a/src/strategies/strategy_c.py b/src/strategies/strategy_c.py new file mode 100644 index 0000000..ed6d95c --- /dev/null +++ b/src/strategies/strategy_c.py @@ -0,0 +1,38 @@ +from api import Api +from src.data_extraction.genius_yield_api_scraper import GeniusYieldAPIScraper +from src.models.candlestick_price_chart import CandlestickPriceChart +from src.utils.logger_utils import LoggerUtils +from src.views.candlestick_price_chart_view import CandlestickPriceChartView + + +# pylint: disable=invalid-name +class strategy_c: + + # pylint: disable=unused-argument + def __init__(self, api_client, config, logger): + logger.info(" > init: strategy_c instance created.") + LoggerUtils.log_warning(logger) + + # Internal state: + self.first_execution_time = None + self.logger = logger + + # Strategy Configuration: + self.asset_pair = config["ASSET_PAIR"] + self.start_time = config["START_TIME"] + self.end_time = config["END_TIME"] + self.bin_interval = config["BIN_INTERVAL"] + + # pylint: disable=unused-argument + def execute(self, api_client : Api, config, logger): + api = GeniusYieldAPIScraper(self.asset_pair, self.start_time, self.end_time, self.bin_interval) + + data = api.fetch_data() + + if data: + parsed_data = api.parse_data(data) + candlestickPriceChart = CandlestickPriceChart(parsed_data) + view = CandlestickPriceChartView(candlestickPriceChart, self.logger) + view.plot() + else: + self.logger.info("🤔") diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/utils/logger_utils.py b/src/utils/logger_utils.py new file mode 100644 index 0000000..ec655c2 --- /dev/null +++ b/src/utils/logger_utils.py @@ -0,0 +1,42 @@ +import logging +from typing import ClassVar + + +class LoggerUtils: + """ + A utility class for logging standardized messages. + + This class provides static methods for logging formatted warning messages + and other utility functions related to logging. + """ + + WARNING_MESSAGE: ClassVar[str] = """ +======================================================================== + + ⚠️ WARNING! ⚠️ + + THIS IS ONLY A PROOF-OF-CONCEPT EXAMPLE STRATEGY IMPLEMENTATION. + + IT IS ONLY INTENDED AS IMPLEMENTATION REFERENCE FOR TRADING STRATEGIES. + + THIS IMPLEMENTATION IS NOT PRODUCTION-READY. + +======================================================================== +""" + + @staticmethod + def log_warning(logger: logging.Logger) -> None: + """ + Log a standardized warning message using the provided logger. + + This method logs a pre-defined warning message to inform users about + the nature and limitations of the example strategy implementation. + + Args: + logger (logging.Logger): The logger instance to use for logging the warning. + + Returns: + None + """ + for line in LoggerUtils.WARNING_MESSAGE.strip().split('\n'): + logger.warning(line) diff --git a/src/utils/market_maker.py b/src/utils/market_maker.py new file mode 100644 index 0000000..2115805 --- /dev/null +++ b/src/utils/market_maker.py @@ -0,0 +1,177 @@ +import logging +import math +from typing import Any, Dict, List + +from api import Api, ApiException + + +class MarketMaker: + """ + Handle buy and sell orders for a specific market. + + This class provides methods to place, cancel, and retrieve buy and sell orders + for a given market using the provided API client. + + Attributes: + api_client (Api): The API client for market interactions. + logger (logging.Logger): Logger for outputting information and errors. + position_size (float): The maximum size of a position in the base asset. + base_asset (str): The base asset of the market. + target_asset (str): The target asset of the market. + market (str): The market identifier (e.g., "BASE_TARGET"). + """ + + def __init__(self, api_client: Api, config: Dict[str, str], logger: logging.Logger): + """ + Initialize the MarketMaker. + + Args: + api_client (Api): The API client for market interactions. + config (Dict[str, str]): Configuration parameters for the market maker. + logger (logging.Logger): Logger for outputting information and errors. + """ + self.api_client = api_client + self.logger = logger + + self.position_size = float(config["POSITION_SIZE_LOVELACES"]) + self.base_asset = config["BASE_ASSET"] + self.target_asset = config["TARGET_ASSET"] + self.market = f"{self.base_asset}_{self.target_asset}" + + def place_buy_order(self, price: float) -> None: + """ + Place a buy order at the specified price. + + Args: + price (float): The price at which to place the buy order. + """ + self.logger.info(" ⚙️ Placing BUY order...") + + try: + balance_available = int(self.api_client.get_balances().get(self.base_asset, 0)) + self.logger.debug(f" > balance_available : {balance_available}") + self.logger.debug(f" > self.position_size: {self.position_size}") + + order_size = min(self.position_size, balance_available) + if not order_size: + self.logger.info(" ⚠️ Insufficient balance to place BUY order! ⚠️") + return + + offered_amount = int(math.floor(order_size)) + + self.logger.info(f" > Place BUY order: {offered_amount} at price {price}...") + response = self.api_client.place_order( + offered_amount=f"{offered_amount}", + offered_token=self.base_asset, + price_token=self.target_asset, + price_amount=f"{int(math.floor(offered_amount / price))}" + ) + self.logger.info(f" > [OK] PLACED NEW BUY ORDER: {response.order_ref}") + # pylint: disable=broad-exception-caught + except Exception as e: + self.logger.error(f" > ⚠️ [FAILED] Could not place BUY order: {str(e)} ⚠️") + self.logger.exception(" > Exception details: ") + + def place_sell_order(self, price: float) -> None: + """ + Place a sell order at the specified price. + + Args: + price (float): The price at which to place the sell order. + """ + self.logger.info(" ⚙️ Placing SELL order...") + + try: + balance_available = int(self.api_client.get_balances().get(self.target_asset, 0)) + self.logger.info(f" > balance_available : {balance_available}") + order_size = min(self.position_size / price, balance_available) + self.logger.info(f" > order_size : {order_size}") + self.logger.info(f" > price : {price}") + if not order_size: + self.logger.info("⚠️ Insufficient balance to place SELL order! ⚠️") + return + + self.logger.info(f" > Place SELL order: {order_size} at price {price}...") + response = self.api_client.place_order( + offered_amount=f"{int(math.floor(order_size))}", + offered_token=self.target_asset, + price_token=self.base_asset, + price_amount=f"{int(math.floor(order_size * price))}" + ) + self.logger.info(f" > [OK] PLACED NEW SELL ORDER: {response.order_ref}") + # pylint: disable=broad-exception-caught + except Exception as e: + self.logger.error(f" > ⚠️ [FAILED] Could not place SELL order: {str(e)} ⚠️") + self.logger.exception(" > Exception details: ") + + def get_buy_orders(self) -> List[Any]: + """ + Get all active buy orders for the current market. + + Returns: + List[Any]: A list of active buy orders. + """ + own_orders = self.api_client.get_own_orders(self.market) + return own_orders.bids + + def get_sell_orders(self) -> List[Any]: + """ + Get all active sell orders for the current market. + + Returns: + List[Any]: A list of active sell orders. + """ + own_orders = self.api_client.get_own_orders(self.market) + return own_orders.asks + + def cancel_buy_orders(self) -> None: + """Cancel all active buy orders for the current market.""" + self.logger.info(" > Cancel all BUY orders...") + self.cancel_orders("bid") + self.logger.info(" > [OK] Canceled all BUY orders.") + + def cancel_sell_orders(self) -> None: + """Cancel all active sell orders for the current market.""" + self.logger.info(" > Cancel all SELL orders...") + self.cancel_orders("ask") + self.logger.info(" > [OK] Canceled all SELL orders.") + + def cancel_orders(self, side: str) -> None: + """ + Cancel all orders of a specific side (buy or sell). + + Args: + side (str): The side of orders to cancel ("ask" for sell, "bid" for buy). + """ + while True: + orders = self.get_sell_orders() if side == "ask" else self.get_buy_orders() + + if not orders: + return + + self.logger.info(f" Remaining {side} orders: {len(orders)}.") + + order = orders[0] + try: + self.logger.info(f" ⚙️ Canceling order: {order.output_reference}") + self.api_client.cancel_order(order.output_reference) + self.logger.info(f" > [OK] Canceled order: {order.output_reference}") + except ApiException as e: + self.logger.error(f" > ⚠️ [FAILED] could not cancel order: {order.output_reference} ⚠️") + self.logger.exception(f" > Exception: {str(e)}") + + def log_orders(self) -> None: + """Log all active orders for the current market.""" + own_orders = self.api_client.get_own_orders(self.market) + + self.logger.info(" ON-CHAIN ORDERS:") + + if not own_orders.asks and not own_orders.bids: + self.logger.info(" > No orders.") + return + + for sell_order in own_orders.asks: + self.logger.info(f" > SELL: {sell_order.output_reference}") + + for buy_order in own_orders.bids: + self.logger.info(f" > BUY: {buy_order.output_reference}") diff --git a/src/views/__init__.py b/src/views/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/views/candlestick_price_chart_view.py b/src/views/candlestick_price_chart_view.py new file mode 100644 index 0000000..817d12d --- /dev/null +++ b/src/views/candlestick_price_chart_view.py @@ -0,0 +1,134 @@ +import logging +from typing import List, Optional + +from src.models.candlestick import Candlestick +from src.models.candlestick_price_chart import CandlestickPriceChart + + +class CandlestickPriceChartView: + """ + A class for visualizing a candlestick price chart. + + This class takes a CandlestickPriceChart and creates an ASCII representation + of the candlestick chart. + + Attributes: + model (CandlestickPriceChart): The model containing the chart data. + logger (Optional[logging.Logger]): A logger for output messages. + """ + + def __init__(self, model: CandlestickPriceChart, logger: Optional[logging.Logger] = None): + """ + Initialize the CandlestickPriceChartView. + + Args: + model (CandlestickPriceChart): The model containing the chart data. + logger (Optional[logging.Logger]): A logger for output messages. If None, print to console. + """ + self.model = model + self.logger = logger + + def plot(self) -> None: + """ + Plot the candlestick chart. + + This method creates an ASCII representation of the candlestick chart + and outputs it using the logger or print function. + """ + if not self.model.candlesticks: + self._log("No data to plot.") + return + + min_price, max_price = self.model.get_price_range() + price_range = max_price - min_price + chart_height = self.model.chart_height + chart_width = self.model.chart_width + + chart = [[" " for _ in range(chart_width)] for _ in range(chart_height)] + + for i, candle in enumerate(self.model.candlesticks): + x = i * 2 + 1 + self._plot_candle(chart, candle, x, min_price, price_range, chart_height) + + self._add_axes(chart, min_price, max_price, chart_height, chart_width) + self._print_chart(chart) + + def _plot_candle(self, chart: List[List[str]], candle: Candlestick, x: int, + min_price: float, price_range: float, chart_height: int) -> None: + """ + Plot a single candlestick on the chart. + + Args: + chart (List[List[str]]): The 2D list representing the chart. + candle (Candlestick): The candlestick to plot. + x (int): The x-coordinate of the candlestick on the chart. + min_price (float): The minimum price in the chart. + price_range (float): The range of prices in the chart. + chart_height (int): The height of the chart. + """ + def price_to_y(price: float) -> int: + return chart_height - 1 - int((price - min_price) / price_range * (chart_height - 1)) + + y_high = price_to_y(candle.high) + y_low = price_to_y(candle.low) + y_open = price_to_y(candle.open) + y_close = price_to_y(candle.close) + + for y in range(y_high, y_low + 1): + if y == y_high or y == y_low: + chart[y][x] = "─" + elif y_open < y_close: # Bullish + chart[y][x] = "│" if y_open <= y <= y_close else "│" + elif y_open > y_close: # Bearish + chart[y][x] = "█" if y_close <= y <= y_open else "│" + else: # Doji + chart[y][x] = "─" + + def _add_axes(self, chart: List[List[str]], min_price: float, max_price: float, + chart_height: int, chart_width: int) -> None: + """ + Add axes and price labels to the chart. + + Args: + chart (List[List[str]]): The 2D list representing the chart. + min_price (float): The minimum price in the chart. + max_price (float): The maximum price in the chart. + chart_height (int): The height of the chart. + chart_width (int): The width of the chart. + """ + for y in range(chart_height): + chart[y][0] = "│" + for x in range(chart_width): + chart[-1][x] = "─" + chart[-1][0] = "└" + + # Add price labels + for i in range(5): + price = min_price + (max_price - min_price) * i / 4 + y = chart_height - 1 - int(i * (chart_height - 1) / 4) + price_str = f"{price:.2f}" + for j, char in enumerate(price_str): + if j < len(chart[y]) - 1: + chart[y][j + 1] = char + + def _print_chart(self, chart: List[List[str]]) -> None: + """ + Print the chart using the logger or print function. + + Args: + chart (List[List[str]]): The 2D list representing the chart. + """ + for row in chart: + self._log("".join(row)) + + def _log(self, message: str) -> None: + """ + Log a message using the logger if available, otherwise print to console. + + Args: + message (str): The message to log or print. + """ + if self.logger: + self.logger.info(message) + else: + print(message) diff --git a/strategies/bollinger_bands_strategy.py b/strategies/bollinger_bands_strategy.py deleted file mode 100644 index 265d021..0000000 --- a/strategies/bollinger_bands_strategy.py +++ /dev/null @@ -1,240 +0,0 @@ -from datetime import datetime -import math -from api import Api, ApiException -import time - -from talipp.indicators import BB - -class bollinger_bands_strategy: - def __init__(self, api_client, CONFIG, logger): - logger.info(" > init: bollinger_bands_strategy instance created.") - - logger.info("========================================================================") - logger.info(" ") - logger.info(" ⚠️ WARNING! ⚠️ ") - logger.info(" ") - logger.info(" THIS IS ONLY A PROOF-OF-CONCEPT EXAMPLE STRATEGY IMPLEMENTATION. ") - logger.info(" ") - logger.info(" IT IS ONLY INTENDED AS IMPLEMENTATION REFERENCE FOR TRADING STRATEGIES.") - logger.info(" ") - logger.info(" THIS IMPLEMENTATION IS NOT PRODUCTION-READY. ") - logger.info(" ") - logger.info("========================================================================") - - # Internal state: - self.last_execution_time = None - self._values = (None, None) - self.api_client : Api = api_client - self.logger = logger - self.initialized = False - self.last_candle = None - - # Strategy configuration: - self.position_size = float(CONFIG["POSITION_SIZE_LOVELACES"]) - self.std_dev_multiplier = float(CONFIG["STD_DEV_MULTIPLIER"]) - self.period = int(CONFIG["PERIOD"]) - self.base_asset = CONFIG["BASE_ASSET"] - self.target_asset = CONFIG["TARGET_ASSET"] - self.market = f"{self.base_asset}_{self.target_asset}" - logger.info(" STRATEGY CONFIGURATION:") - logger.info(f" > base_asset : {self.base_asset}") - logger.info(f" > target_asset : {self.target_asset}") - logger.info(f" > market : {self.market}") - logger.info(f" > position_size : {self.position_size}") - logger.info(f" > std_dev_multiplier : {self.std_dev_multiplier}") - logger.info(f" > period : {self.period}") - - # Create the BB strategy instance with the config: - self.bb = BB(self.period, self.std_dev_multiplier) - - def place_buy_order(self, api_client, logger, price): - logger.info(" ⚙️ Placing BUY order...") - - try: - balance_available = int(api_client.get_balances().get(self.base_asset, 0)) - logger.debug(f" > balance_available : {balance_available}") - logger.debug(f" > self.position_size: {self.position_size}") - - order_size = min(self.position_size, balance_available) - if not order_size: - logger.info(" ⚠️ Insufficient balance to place BUY order! ⚠️") - return - - offered_amount = int(math.floor(order_size)) - - logger.info(f" > Place BUY order: {offered_amount} at price {price}...") - response = api_client.place_order( - offered_amount=f"{offered_amount}", - offered_token=self.base_asset, - price_token=self.target_asset, - price_amount=f"{int(math.floor(offered_amount / price))}" - ) - logger.info(f" > [OK] PLACED NEW BUY ORDER: {response.order_ref}") - except: - logger.error(" > ⚠️ [FAILED] Could not place BUY order. ⚠️") - logger.exception(f" > Exception! ") - - def place_sell_order(self, api_client, logger, price): - logger.info(" ⚙️ Placing SELL order...") - - try: - balance_available = int(api_client.get_balances().get(self.target_asset, 0)) - logger.info(f" > balance_available : {balance_available}") - order_size = min(self.position_size / price, balance_available) - logger.info(f" > order_size : {order_size}") - logger.info(f" > price : {price}") - if not order_size: - logger.info("⚠️ Insufficient balance to place SELL order! ⚠️") - return - - logger.info(f" > Place SELL order: {order_size} at price {price}...") - response = api_client.place_order( - offered_amount=f"{int(math.floor(order_size))}", - offered_token=self.target_asset, - price_token=self.base_asset, - price_amount=f"{int(math.floor(order_size * price))}" - ) - logger.info(f" > [OK] PLACED NEW SELL ORDER: {response.order_ref}") - except: - logger.error(f" > ⚠️ [FAILED] Could not place SELL order. ⚠️") - logger.exception(f" > Exception! ") - - def get_buy_orders(self): - own_orders = self.api_client.get_own_orders(self.market) - return own_orders.bids - - def get_sell_orders(self): - own_orders = self.api_client.get_own_orders(self.market) - return own_orders.asks - - def cancel_buy_orders(self): - self.logger.info(" > Cancel all BUY orders...") - self.cancel_orders("bid") - self.logger.info(" > [OK] Canceled all BUY orders.") - - def cancel_sell_orders(self): - self.logger.info(" > Cancel all SELL orders...") - self.cancel_orders("ask") - self.logger.info(" > [OK] Canceled all SELL orders.") - - def cancel_orders(self, side): - while True: - orders = [] - own_orders = self.api_client.get_own_orders(self.market) - if (side == "ask"): - orders = own_orders.asks - else: - orders = own_orders.bids - - if len(orders) == 0: - return - else: - self.logger.info(f" Remaining {side} orders: {len(orders)}.") - - order = orders[0] - try: - self.logger.info(f" ⚙️ Canceling order: {order.output_reference}") - self.api_client.cancel_order(order.output_reference) - self.logger.info(f" > [OK] Canceled order: {order.output_reference}") - except ApiException: - self.logger.error(f" > ⚠️ [FAILED] could not cancel order: {order.output_reference} ⚠️") - self.logger.exception(f" > Exception! ") - - def process_candle(self, candle): - if self.initialized: - self.logger.info(f" > processsing candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}") - else: - self.logger.info(f" > processsing init candle - timestamp: {candle.timestamp} - base_close: {candle.base_close}") - - if (not self.last_candle == None) and (self.last_candle.timestamp == candle.timestamp): - self.logger.info(f" > Candle has already been processsed. Nothing to do.") - return - - self.last_candle = candle - - # Feed the technical indicator. - value = float(candle.base_close) - self.bb.add(value) - - # Keep a small window of values to check if there is a crossover. - self._values = (self._values[-1], value) - - if len(self.bb) < 2 or self.bb[-1] == None or self.bb[-2] == None: - self.logger.info(f" BOLLINGER BANDS: Initializing... ⚙️ ⏳ ") - self.logger.info(f" > Upper band: Not available.") - self.logger.info(f" > Lower band: Not available.") - return - - self.logger.info(f" BOLLINGER BANDS: ") - self.logger.info(f" > Upper band: {self.bb[-1].ub}") - self.logger.info(f" > Lower band: {self.bb[-1].lb}") - - if self.initialized == False: - self.logger.info(f" -> Initializaion phase. Do not place orders yet.") - return - - self.place_buy_order(self.api_client, self.logger, candle.base_close) - - # Price moved below lower band ? - if self._values[-2] >= self.bb[-2].lb and self._values[-1] < self.bb[-1].lb: - self.logger.info(f" -> Price moved below the lower band -> BUY! 🛒 🛒 🛒 ") - self.cancel_sell_orders() - if len(self.get_buy_orders()) > 0: - self.logger.info(" > Already placed BUY order. Nothing to do.") - else: - self.place_buy_order(self.api_client, self.logger, candle.base_close) - # Price moved above upper band ? - elif self._values[-2] <= self.bb[-2].ub and self._values[-1] > self.bb[-1].ub: - self.logger.info(f" -> Price moved above the upper band -> SELL! 💲 💲 💲 ") - self.cancel_buy_orders() - if len(self.get_sell_orders()) > 0: - self.logger.info(" > Already placed SELL order. Nothing to do.") - else: - self.place_sell_order(self.api_client, self.logger, candle.base_low) - - self.log_orders() - - def log_orders(self): - own_orders = self.api_client.get_own_orders(self.market) - - self.logger.info(" ON-CHAIN ORDERS:") - - if (len(own_orders.asks) + len(own_orders.bids)) == 0: - self.logger.info(f" > No orders.") - return - - for sell_order in own_orders.asks: - self.logger.info(f" > SELL: {sell_order.output_reference}") - - for buy_order in own_orders.bids: - self.logger.info(f" > BUY: {buy_order.output_reference} ") - - def execute(self, api_client : Api, CONFIG, logger): - current_time = datetime.now() - - if self.last_execution_time is None: - logger.info("Executing for the first time -> initialize.") - candles = api_client.get_price_history(self.market, resolution="1m", sort="asc", limit=self.period*5) - for candle in candles[:-1]: - self.logger.info(f"--------------------------------------------------------------------------------") - self.process_candle(candle) - time.sleep(1) - logger.info(" > [OK] Initialized.") - logger.info("========================================================================") - self.initialized = True - self.last_candle=None - else: - time_since_last_execution = (current_time - self.last_execution_time).total_seconds() - logger.info(f"Last executed: {self.last_execution_time}") - logger.info(f"Seconds since last execution: {time_since_last_execution} seconds") - - self.last_execution_time = current_time # Update last execution time - self.initialized = True - - try: - get_market_price = api_client.get_market_price(self.market) - candle=get_market_price[0] - self.process_candle(candle) - except: - logger.error(f" > ⚠️ [FAILED] could not process candle ⚠️") - logger.exception(f" > Exception! ") diff --git a/strategies/strategy_b.py b/strategies/strategy_b.py deleted file mode 100644 index df8e518..0000000 --- a/strategies/strategy_b.py +++ /dev/null @@ -1,41 +0,0 @@ -from datetime import datetime - -class strategy_b: - def __init__(self, client, CONFIG, logger): - self.first_execution_time = None - self.client = client - self.CONFIG = CONFIG - self.logger = logger - logger.info("Strategy B instance created") - - logger.info("========================================================================") - logger.info(" ") - logger.info(" ⚠️ WARNING! ⚠️ ") - logger.info(" ") - logger.info(" THIS IS ONLY A PROOF-OF-CONCEPT EXAMPLE STRATEGY IMPLEMENTATION. ") - logger.info(" ") - logger.info(" IT IS ONLY INTENDED AS IMPLEMENTATION REFERENCE FOR TRADING STRATEGIES.") - logger.info(" ") - logger.info(" THIS IMPLEMENTATION IS NOT PRODUCTION-READY. ") - logger.info(" ") - logger.info("========================================================================") - - def execute(self, client, CONFIG, logger): - current_time = datetime.now() - if self.first_execution_time is None: - self.first_execution_time = current_time - - # Calculate the time difference in seconds - time_since_first_execution = (current_time - self.first_execution_time).total_seconds() - - # Toggle behavior every 30 seconds - if int(time_since_first_execution // 30) % 2 == 0: - self.behavior_a() - else: - self.behavior_b() - - def behavior_a(self): - self.logger.info("Executing Strategy B: Behavior A 🍎") - - def behavior_b(self): - self.logger.info("Executing Strategy B: Behavior B 🍌")