From 2c5b7da8165e3f8b90748e8ced1de0e3e8d52309 Mon Sep 17 00:00:00 2001 From: lazydancer Date: Sun, 22 Dec 2024 17:33:11 -0500 Subject: Initial --- backend/src/arbitrage.py | 141 ++++++++++++++++++ backend/src/esi/api.py | 243 +++++++++++++++++++++++++++++++ backend/src/esi/esi_oauth_web.py | 87 +++++++++++ backend/src/esi/refresh_token.py | 43 ++++++ backend/src/esi/shared_flow.py | 144 ++++++++++++++++++ backend/src/esi/validate_jwt.py | 97 +++++++++++++ backend/src/graph.py | 305 +++++++++++++++++++++++++++++++++++++++ backend/src/loader.py | 186 ++++++++++++++++++++++++ backend/src/main.py | 94 ++++++++++++ backend/src/manager.py | 104 +++++++++++++ backend/src/market.py | 203 ++++++++++++++++++++++++++ backend/src/pathfinder.py | 145 +++++++++++++++++++ backend/src/route.py | 208 ++++++++++++++++++++++++++ 13 files changed, 2000 insertions(+) create mode 100644 backend/src/arbitrage.py create mode 100644 backend/src/esi/api.py create mode 100644 backend/src/esi/esi_oauth_web.py create mode 100644 backend/src/esi/refresh_token.py create mode 100644 backend/src/esi/shared_flow.py create mode 100644 backend/src/esi/validate_jwt.py create mode 100644 backend/src/graph.py create mode 100644 backend/src/loader.py create mode 100644 backend/src/main.py create mode 100644 backend/src/manager.py create mode 100644 backend/src/market.py create mode 100644 backend/src/pathfinder.py create mode 100644 backend/src/route.py (limited to 'backend/src') diff --git a/backend/src/arbitrage.py b/backend/src/arbitrage.py new file mode 100644 index 0000000..0991bbc --- /dev/null +++ b/backend/src/arbitrage.py @@ -0,0 +1,141 @@ +import logging +from dataclasses import dataclass +from typing import Optional, List + +from market import Order + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +TAX_RATE = 0.08 +PROFIT_THRESHOLD = 100_000 +MAX_CAPITAL = 100_000_000 +SNIPE_PROFIT_THRESHOLD = 20_000_000 + +@dataclass +class Trade: + from_station: int + to_station: int + from_order_id: int + to_order_id: int + item_name: str + type_id: int + item_cargo_volume: float + from_price: float + to_price: float + quantity: int + cargo: float + gross_profit: float + net_profit: Optional[float] = None + +def arbitrage(orders: List[Order], cargo_space: int) -> List[Trade]: + sell_orders = sorted( + (order for order in orders if not order.is_buy_order), + key=lambda x: x.price + ) + buy_orders = sorted( + (order for order in orders if order.is_buy_order), + key=lambda x: x.price, reverse=True + ) + + logger.info("Initial sell orders: %d, buy orders: %d", len(sell_orders), len(buy_orders)) + + sell_orders = filter_orders_for_cargo(sell_orders, cargo_space) + buy_orders = filter_orders_for_cargo(buy_orders, cargo_space) + logger.info("After cargo filter - sell orders: %d, buy orders: %d", len(sell_orders), len(buy_orders)) + + sell_orders = filter_orders_for_capital_risk(sell_orders, MAX_CAPITAL) + buy_orders = filter_orders_for_capital_risk(buy_orders, MAX_CAPITAL) + logger.info("After capital risk filter - sell orders: %d, buy orders: %d", len(sell_orders), len(buy_orders)) + + trades = create_trades(sell_orders, buy_orders) + return trades + +def create_trades(sell_orders: List[Order], buy_orders: List[Order]) -> List[Trade]: + trades = [] + grouped_sell_orders = {} + grouped_buy_orders = {} + + for order in sell_orders: + grouped_sell_orders.setdefault(order.type_id, []).append(order) + for order in buy_orders: + grouped_buy_orders.setdefault(order.type_id, []).append(order) + + for type_id in grouped_sell_orders.keys() & grouped_buy_orders.keys(): + for sell_order in grouped_sell_orders[type_id]: + for buy_order in grouped_buy_orders[type_id]: + effective_buy_price = buy_order.price * (1 - TAX_RATE) + if effective_buy_price < sell_order.price: + continue + + quantity = min(sell_order.volume_remain, buy_order.volume_remain) + gross_profit = quantity * (effective_buy_price - sell_order.price) + + if gross_profit < PROFIT_THRESHOLD: + continue + + cargo_volume = quantity * sell_order.item_cargo_volume + + trade = Trade( + from_station=sell_order.location_id, + to_station=buy_order.location_id, + from_order_id=sell_order.order_id, + to_order_id=buy_order.order_id, + item_name=sell_order.item_name, + type_id=sell_order.type_id, + item_cargo_volume=sell_order.item_cargo_volume, + from_price=sell_order.price, + to_price=buy_order.price, + quantity=quantity, + cargo=cargo_volume, + gross_profit=gross_profit + ) + trades.append(trade) + return trades + +def filter_orders_for_cargo(orders: List[Order], cargo_space: int) -> List[Order]: + filtered_orders = [] + for order in orders: + item_volume = order.item_cargo_volume + if item_volume == 0: + continue + max_quantity = int(cargo_space // item_volume) + volume_remain = min(order.volume_remain, max_quantity) + if volume_remain > 0: + order = Order(**vars(order)) + order.volume_remain = volume_remain + filtered_orders.append(order) + return filtered_orders + +def filter_orders_for_capital_risk(orders: List[Order], max_amount: int) -> List[Order]: + filtered_orders = [] + for order in orders: + price = order.price + if price == 0: + continue + max_quantity = int(max_amount // price) + volume_remain = min(order.volume_remain, max_quantity) + if volume_remain > 0: + order = Order(**vars(order)) + order.volume_remain = volume_remain + filtered_orders.append(order) + return filtered_orders + +def snipe(orders: List[Order], cargo_space: int, current_region: int) -> Optional[Trade]: + trades = arbitrage(orders, cargo_space) + + order_regions = {order.order_id: order.region_id for order in orders} + + region_trades = [ + trade for trade in trades + if order_regions.get(trade.from_order_id) == current_region + ] + + region_trades.sort(key=lambda x: x.gross_profit, reverse=True) + + logger.info("Reviewing potential snipes for region %d: %d trades found", current_region, len(region_trades)) + + if region_trades and region_trades[0].gross_profit > SNIPE_PROFIT_THRESHOLD: + return region_trades[0] + else: + return None diff --git a/backend/src/esi/api.py b/backend/src/esi/api.py new file mode 100644 index 0000000..862b4bf --- /dev/null +++ b/backend/src/esi/api.py @@ -0,0 +1,243 @@ +""" +A simplified API for interacting with the EVE Online ESI API. +""" + +import datetime +import json +import logging +import os +import queue +import threading +import time +from pathlib import Path +from typing import Any + +import requests + +from .refresh_token import get_access_token_from_refresh_token + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Constants +ESI_BASE_URL = "https://esi.evetech.net/latest" +DATASOURCE = "tranquility" +USER_AGENT_HEADER = {'User-Agent': 'Haul (Hyperspace Asset Unloading and Loading)'} +CACHE_DIR = Path(__file__).resolve().parent.parent / 'cache' +AUTH_FILE = CACHE_DIR / 'auth.json' +ERROR_LIMIT_THRESHOLD = 50 # ESI Error limit threshold +REQUEST_RATE_LIMIT = 0.05 # Time in seconds between requests to avoid rate limiting + + +def get_location() -> int: + """ + Gets the current location of the character. + + Returns: + The location ID of the character (station ID or solar system ID). + """ + character_id = os.getenv('character_id') + if not character_id: + logger.error("Character ID not found in environment variables.") + raise EnvironmentError("Character ID not set in environment variables.") + + url = f"{ESI_BASE_URL}/characters/{character_id}/location/" + + headers = { + "Accept": "application/json", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {access_token()}", + **USER_AGENT_HEADER, + } + + params = { + "datasource": DATASOURCE, + } + + response = requests.get(url, params=params, headers=headers) + logger.debug(f"Request URL: {response.url}") + logger.debug(f"Response Status Code: {response.status_code}") + + if response.status_code == 401: + logger.info("Access token expired, refreshing token.") + access_token(force_refresh=True) + headers["Authorization"] = f"Bearer {access_token()}" + response = requests.get(url, params=params, headers=headers) + + if int(response.headers.get("X-Esi-Error-Limit-Remain", 100)) < ERROR_LIMIT_THRESHOLD: + logger.warning("Approaching ESI Error Limit.") + raise Exception("ESI Error Limit approaching threshold.") + + response.raise_for_status() + data: dict[str, Any] = response.json() + + location_id = data.get("station_id") or data.get("solar_system_id") + if not location_id: + logger.error("Location ID not found in response.") + raise ValueError("Unable to determine character's location.") + + logger.info(f"Character location ID: {location_id}") + return location_id + + +def set_waypoints(waypoints: list[int]) -> None: + """ + Sets the waypoints for the character's autopilot. + + Args: + waypoints: A list of waypoint IDs to set. + """ + if not waypoints: + logger.warning("No waypoints provided.") + return + + url = f"{ESI_BASE_URL}/ui/autopilot/waypoint/" + + headers = { + "Accept": "application/json", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {access_token()}", + **USER_AGENT_HEADER, + } + + for index, waypoint in enumerate(waypoints): + params = { + "add_to_beginning": False, + "clear_other_waypoints": index == 0, + "destination_id": waypoint, + "datasource": DATASOURCE, + } + + try: + response = requests.post(url, params=params, headers=headers) + response.raise_for_status() + logger.info(f"Waypoint {waypoint} set successfully.") + time.sleep(1) # Rate limiting + except requests.HTTPError as e: + logger.error(f"Failed to set waypoint {waypoint}: {e}") + continue + + +def open_market_window(type_id: int) -> None: + """ + Opens the market window for a specific item type in the game client. + + Args: + type_id: The type ID of the item. + """ + url = f"{ESI_BASE_URL}/ui/openwindow/marketdetails/" + + headers = { + "Accept": "application/json", + "Cache-Control": "no-cache", + "Authorization": f"Bearer {access_token()}", + **USER_AGENT_HEADER, + } + + params = { + "type_id": type_id, + "datasource": DATASOURCE, + } + + try: + response = requests.post(url, params=params, headers=headers) + response.raise_for_status() + logger.info(f"Market window opened for type ID {type_id}.") + except requests.HTTPError as e: + logger.error(f"Failed to open market window for type ID {type_id}: {e}") + + +def get_region_orders(region_id: int) -> tuple[list[dict[str, Any]], str]: + """ + Retrieves all market orders for a given region. + + Args: + region_id: The ID of the region. + + Returns: + A tuple containing the list of orders and the expiry time. + """ + def request_orders(region: int, page: int, q: queue.Queue) -> None: + url = f"{ESI_BASE_URL}/markets/{region}/orders/" + params = { + 'datasource': DATASOURCE, + 'order_type': 'all', + 'page': page, + } + headers = USER_AGENT_HEADER + + try: + response = requests.get(url, params=params, headers=headers) + response.raise_for_status() + logger.debug(f"Retrieved page {page} for region {region}.") + q.put(response) + except requests.HTTPError as e: + logger.error(f"Error fetching page {page} for region {region}: {e}") + + order_queue = queue.Queue() + request_orders(region_id, 1, order_queue) # Get the first page + first_response = order_queue.get() + pages = int(first_response.headers.get('X-Pages', '1')) + expires = first_response.headers.get('Expires', '') + + if pages > 1: + threads = [] + for page in range(2, pages + 1): + thread = threading.Thread(target=request_orders, args=(region_id, page, order_queue)) + threads.append(thread) + thread.start() + time.sleep(REQUEST_RATE_LIMIT) # Rate limiting + + for thread in threads: + thread.join() + + results: list[dict[str, Any]] = first_response.json() + while not order_queue.empty(): + response = order_queue.get() + results.extend(response.json()) + + # Filter out orders with zero remaining volume + results = [order for order in results if order.get('volume_remain', 0) > 0] + + # Add region ID to each order + for order in results: + order['region_id'] = region_id + + logger.info(f"Retrieved {len(results)} orders for region {region_id}.") + return results, expires + + +def access_token(force_refresh: bool = False) -> str: + """ + Retrieves an active access token, refreshing it if necessary. + + Args: + force_refresh: If True, forces a refresh of the access token. + + Returns: + The active access token as a string. + """ + try: + with AUTH_FILE.open("r", encoding="utf-8") as f: + auth: dict[str, Any] = json.load(f) + except FileNotFoundError: + logger.error(f"Authentication file not found at {AUTH_FILE}.") + raise FileNotFoundError("Authentication file is missing.") + + expires_at = auth.get("expires_at", 0) + current_time = datetime.datetime.utcnow().timestamp() + if not force_refresh and expires_at > current_time: + return auth.get("access_token", "") + + logger.info("Access token expired or force refresh requested, refreshing token.") + new_auth = get_access_token_from_refresh_token(auth["refresh_token"]) + + # Update expiry time (assuming token is valid for 20 minutes) + new_auth["expires_at"] = current_time + 1200 + + with AUTH_FILE.open("w", encoding="utf-8") as f: + json.dump(new_auth, f) + + logger.info("Access token refreshed successfully.") + return new_auth["access_token"] diff --git a/backend/src/esi/esi_oauth_web.py b/backend/src/esi/esi_oauth_web.py new file mode 100644 index 0000000..5440bf9 --- /dev/null +++ b/backend/src/esi/esi_oauth_web.py @@ -0,0 +1,87 @@ +""" +Python 3 Web Application OAuth 2.0 Example. + +This script demonstrates how to perform the OAuth 2.0 authorization code flow +for a web-based application using the EVE Online SSO. + +Prerequisites: + * Create an SSO application at https://developers.eveonline.com with the + scope "esi-characters.read_blueprints.v1" and the callback URL + "https://localhost/callback/". Note: Never use localhost as a callback + in production applications. + * Have a Python 3 environment available (possibly by using a virtual + environment: https://virtualenv.pypa.io/en/stable/). + * Install the required packages by running: + pip install -r requirements.txt + +To run this example, make sure you have completed the prerequisites and then +run the following command: + +>>> python esi_oauth_web.py + +Then follow the prompts. +""" + +import base64 +import os +from typing import Optional + +from dotenv import load_dotenv + +from shared_flow import ( + handle_sso_token_response, + print_auth_url, + send_token_request, +) + + +def main() -> None: + """ + Guides you through a local example of the OAuth 2.0 web flow. + """ + print( + "This program will take you through an example OAuth 2.0 flow " + "that you should be using if you are building a web-based " + "application. Follow the prompts and enter the information requested." + ) + + client_id: Optional[str] = os.getenv('CLIENT_ID') + if not client_id: + raise ValueError("CLIENT_ID not found in environment variables.") + + print_auth_url(client_id) + + auth_code: str = input('Copy the "code" query parameter and enter it here: ') + + # Basic auth can be handled by the requests library with the auth parameter, + # but we're constructing it manually for educational purposes. + secret_key: Optional[str] = os.getenv('SECRET_KEY') + if not secret_key: + raise ValueError("SECRET_KEY not found in environment variables.") + + user_pass: str = f"{client_id}:{secret_key}" + basic_auth: str = base64.urlsafe_b64encode(user_pass.encode('utf-8')).decode() + auth_header: str = f"Basic {basic_auth}" + + form_values: dict[str, str] = { + "grant_type": "authorization_code", + "code": auth_code, + } + + headers: dict[str, str] = {"Authorization": auth_header} + + print( + "\nThe following request uses basic authentication, as a web-based " + "app you should do the same." + ) + + input("\nPress Enter to continue...") + + res = send_token_request(form_values, add_headers=headers) + + handle_sso_token_response(res) + + +if __name__ == "__main__": + load_dotenv() + main() diff --git a/backend/src/esi/refresh_token.py b/backend/src/esi/refresh_token.py new file mode 100644 index 0000000..0b3e4a3 --- /dev/null +++ b/backend/src/esi/refresh_token.py @@ -0,0 +1,43 @@ +import base64 +import datetime +import os + +from .shared_flow import send_token_request + +def get_access_token_from_refresh_token(refresh_token: str) -> dict[str, any]: + """ + Obtains a new access token using a refresh token. + + This function uses the provided refresh token to request a new access token + from the authentication server. It updates the expiration time of the token. + + Args: + refresh_token: The refresh token used to obtain a new access token. + + Returns: + A dictionary containing the new access token data, including 'expires_at'. + """ + client_id = os.getenv('CLIENT_ID') + secret_key = os.getenv('SECRET_KEY') + + if not client_id or not secret_key: + raise EnvironmentError("CLIENT_ID and SECRET_KEY must be set in environment variables.") + + user_pass = f"{client_id}:{secret_key}" + basic_auth = base64.urlsafe_b64encode(user_pass.encode('utf-8')).decode() + auth_header = f"Basic {basic_auth}" + + form_values = { + "grant_type": "refresh_token", + "refresh_token": refresh_token, + } + + headers = {"Authorization": auth_header} + + response = send_token_request(form_values, add_headers=headers) + data = response.json() + + current_timestamp = int(datetime.datetime.now().timestamp()) + data["expires_at"] = current_timestamp + data.get("expires_in", 0) + + return data diff --git a/backend/src/esi/shared_flow.py b/backend/src/esi/shared_flow.py new file mode 100644 index 0000000..1d89fb1 --- /dev/null +++ b/backend/src/esi/shared_flow.py @@ -0,0 +1,144 @@ +""" +Contains shared OAuth 2.0 flow functions for EVE Online examples. + +This module contains shared functions used between the two different OAuth 2.0 +flows recommended for web-based and mobile/desktop applications. The functions +found here are used by the OAuth 2.0 examples contained in this project. +""" + +import requests + +from .validate_jwt import validate_eve_jwt + + +def print_auth_url(client_id: str, code_challenge: str = "") -> None: + """ + Prints the URL to redirect users to for OAuth authorization. + + Args: + client_id: The client ID of an EVE SSO application. + code_challenge: A PKCE code challenge (optional). + """ + base_auth_url = "https://login.eveonline.com/v2/oauth/authorize/" + params = { + "response_type": "code", + "redirect_uri": "https://localhost/callback/", + "client_id": client_id, + "state": "unique-state", + } + + # Define the required scopes + scopes = [ + "esi-assets.read_assets.v1", + "esi-location.read_location.v1", + "esi-location.read_online.v1", + "esi-ui.open_window.v1", + "esi-ui.write_waypoint.v1", + ] + scope_param = "%20".join(scopes) + params["scope"] = scope_param + + if code_challenge: + params["code_challenge"] = code_challenge + params["code_challenge_method"] = "S256" + + # Construct the full authorization URL + request = requests.Request('GET', base_auth_url, params=params) + prepared_request = request.prepare() + full_auth_url = prepared_request.url + + print( + "\nOpen the following link in your browser:\n\n" + f"{full_auth_url}\n\n" + "Once you have logged in as a character, you will be redirected to " + "https://localhost/callback/." + ) + + +def send_token_request(form_values: dict[str, str], add_headers: dict[str, str] = {}) -> requests.Response: + """ + Sends a request for an authorization token to the EVE SSO. + + Args: + form_values: A dictionary containing the form-encoded values to send with the request. + add_headers: A dictionary containing additional headers to send (optional). + + Returns: + A requests.Response object containing the server's response. + """ + headers = { + "Content-Type": "application/x-www-form-urlencoded", + "Host": "login.eveonline.com", + } + headers.update(add_headers) + + response = requests.post( + "https://login.eveonline.com/v2/oauth/token", + data=form_values, + headers=headers, + ) + + print( + f"Request sent to URL {response.url} with headers {headers} and form values: {form_values}\n" + ) + response.raise_for_status() + + return response + + +def handle_sso_token_response(sso_response: requests.Response) -> None: + """ + Handles the authorization code response from the EVE SSO. + + Args: + sso_response: A requests.Response object from the EVE SSO /v2/oauth/token endpoint. + """ + if sso_response.status_code == 200: + data = sso_response.json() + access_token = data.get("access_token") + + print("\nVerifying access token JWT...") + + jwt = validate_eve_jwt(access_token) + character_id = jwt["sub"].split(":")[2] + character_name = jwt["name"] + blueprint_url = f"https://esi.evetech.net/latest/characters/{character_id}/blueprints/" + + print( + f"\nSuccess! Here is the payload received from the EVE SSO: {data}\n" + "You can use the access token to make an authenticated request to " + f"{blueprint_url}" + ) + + input("\nPress Enter to have this program make the request for you:") + + headers = { + "Authorization": f"Bearer {access_token}" + } + + res = requests.get(blueprint_url, headers=headers) + print( + f"\nMade request to {blueprint_url} with headers: {res.request.headers}" + ) + res.raise_for_status() + + blueprints = res.json() + print(f"\n{character_name} has {len(blueprints)} blueprints.") + else: + print( + "\nSomething went wrong! Please re-read the comments at the top of this " + "file and ensure you completed all the prerequisites, then try again. " + "Here's some debug info to help you out:" + ) + request_info = sso_response.request + print( + f"\nSent request with URL: {request_info.url}\n" + f"Body: {request_info.body}\n" + f"Headers: {request_info.headers}" + ) + print(f"\nSSO response code is: {sso_response.status_code}") + try: + error_data = sso_response.json() + print(f"\nSSO response JSON is: {error_data}") + except ValueError: + print("\nNo JSON response received from the SSO.") diff --git a/backend/src/esi/validate_jwt.py b/backend/src/esi/validate_jwt.py new file mode 100644 index 0000000..054c367 --- /dev/null +++ b/backend/src/esi/validate_jwt.py @@ -0,0 +1,97 @@ +""" +Validates a given JWT access token originating from the EVE SSO. + +Prerequisites: + * Have a Python 3 environment available (possibly using a virtual environment). + * Run pip install -r requirements.txt with this directory as your root. + +This script can be run by executing: + +>>> python validate_jwt.py + +and providing a JWT access token that you have retrieved from the EVE SSO. +""" + +import sys +from typing import Any + +import requests +from jose import jwt +from jose.exceptions import ExpiredSignatureError, JWTError + +SSO_META_DATA_URL = "https://login.eveonline.com/.well-known/oauth-authorization-server" +JWK_ALGORITHM = "RS256" +JWK_ISSUERS = ("login.eveonline.com", "https://login.eveonline.com") +JWK_AUDIENCE = "EVE Online" + + +def validate_eve_jwt(token: str) -> dict[str, Any]: + """ + Validates a JWT access token retrieved from the EVE SSO. + + Args: + token: A JWT access token originating from the EVE SSO. + + Returns: + The contents of the validated JWT access token. + + Raises: + RuntimeError: If required data is missing from the SSO endpoints. + jose.exceptions.JWTError: If the token is invalid or expired. + """ + response = requests.get(SSO_META_DATA_URL) + response.raise_for_status() + metadata = response.json() + + jwks_uri = metadata.get("jwks_uri") + if not jwks_uri: + raise RuntimeError(f"Invalid data received from the SSO metadata endpoint: {metadata}") + + response = requests.get(jwks_uri) + response.raise_for_status() + jwks_data = response.json() + + jwk_keys = jwks_data.get("keys") + if not jwk_keys: + raise RuntimeError(f"Invalid data received from the JWKS endpoint: {jwks_data}") + + jwk = next((item for item in jwk_keys if item["alg"] == JWK_ALGORITHM), None) + if not jwk: + raise RuntimeError(f"No JWK found with algorithm {JWK_ALGORITHM}") + + # Decode and validate the token + contents = jwt.decode( + token=token, + key=jwk, + algorithms=[JWK_ALGORITHM], + issuer=JWK_ISSUERS, + audience=JWK_AUDIENCE, + ) + return contents + + +def main() -> None: + """ + Entry point of the script. Prompts the user for a JWT token and validates it. + """ + token = input("Enter an access token to validate: ") + + try: + token_contents = validate_eve_jwt(token) + except ExpiredSignatureError: + print("The JWT token has expired.") + sys.exit(1) + except JWTError as e: + print(f"The JWT token is invalid: {e}") + sys.exit(1) + except RuntimeError as e: + print(f"An error occurred: {e}") + sys.exit(1) + else: + print("\nThe contents of the access token are:") + for key, value in token_contents.items(): + print(f"{key}: {value}") + + +if __name__ == "__main__": + main() diff --git a/backend/src/graph.py b/backend/src/graph.py new file mode 100644 index 0000000..31eef80 --- /dev/null +++ b/backend/src/graph.py @@ -0,0 +1,305 @@ +import logging +import os +import pickle +import networkx as nx +from math import sqrt, log +from pathlib import Path +from typing import Any, Optional + +from loader import load_map_data, load_star_gate_connections + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +PATHS_FILE = Path(__file__).resolve().parent.parent / 'cache' / 'path_cache.pkl' + + +class NavigationGraph: + """ + Represents the navigation graph for stations and solar systems, + allowing computation of shortest paths based on time and risk. + """ + + def __init__(self, ship: dict[str, Any]): + """ + Initializes the NavigationGraph. + + Args: + ship: A dictionary containing ship parameters like + max warp speed, max subwarp speed, etc. + """ + self.graph = nx.Graph() + self.stations, self.solar_systems = load_map_data() + self._add_intra_system_connections() + self._load_star_gate_connections() + self._add_time_and_risk(ship) + self.paths = self._load_paths() + + def shortest_path(self, source: int, target: int) -> Optional[list[int]]: + """ + Returns the shortest path between two stations. + + Args: + source: The starting station ID. + target: The target station ID. + + Returns: + A list of station IDs representing the shortest path, + or None if no path exists. + """ + try: + path = nx.shortest_path(self.graph, source, target, weight="weight") + return path + except nx.NetworkXNoPath: + logger.warning(f"No path found between {source} and {target}") + return None + + def shortest_path_length(self, source: int, target: int) -> Optional[float]: + """ + Returns the length of the shortest path between two stations. + + Args: + source: The starting station ID. + target: The target station ID. + + Returns: + The length of the shortest path, or None if no path exists. + """ + try: + return self.paths[source][target] + except KeyError: + logger.warning(f"No path length found between {source} and {target}") + return None + + def formatted_route(self, path: list[int]) -> list[dict[str, Any]]: + """ + Formats the route for the ESI API and frontend. + + Args: + path: A list of station IDs representing the path. + + Returns: + A list of dictionaries containing location information. + """ + combined_path = self._combine_warp_gates(path) + formatted_route = [] + + for index, location_id in enumerate(combined_path): + location_info = self._get_location_info(location_id, index) + if location_info is not None: + formatted_route.append(location_info) + + return formatted_route + + def _get_location_info(self, location_id: int, index: int) -> Optional[dict[str, Any]]: + """ + Retrieves location information for a given ID. + + Args: + location_id: The ID of the location (station or system). + index: The position in the route. + + Returns: + A dictionary with location details, or None if not found. + """ + if location_id in self.stations: + station_data = self.stations[location_id] + solar_system = self.solar_systems.get(station_data['solar_system_id'], {}) + return { + "id": index, + "location_id": location_id, + "station_id": location_id, + "solar_system_id": station_data['solar_system_id'], + "location_type": "station", + "location": f"{station_data['region_name']} - {solar_system.get('item_name', '')} - {station_data['item_name']}", + "actions": [], + } + elif location_id in self.solar_systems: + solar_system_data = self.solar_systems[location_id] + return { + "id": index, + "location_id": location_id, + "solar_system_id": location_id, + "location_type": "system", + "location": f"{solar_system_data['region_name']} - {solar_system_data['item_name']}", + "actions": [], + } + else: + logger.warning(f"Location ID {location_id} not found in stations or solar systems") + return None + + def _combine_warp_gates(self, path: list[int]) -> list[int]: + """ + Combines warp gates in the path to reduce redundant entries. + + Args: + path: A list of station IDs. + + Returns: + A list of location IDs with combined warp gates. + """ + combined_path = [] + + for i, (current_id, next_id) in enumerate(zip(path, path[1:])): + current_station = self.stations.get(current_id) + next_station = self.stations.get(next_id) + + if current_station and current_station['is_station']: + combined_path.append(current_id) + elif (current_station and next_station and + not current_station['is_station'] and + not next_station['is_station'] and + current_station['solar_system_id'] == next_station['solar_system_id']): + combined_path.append(current_station['solar_system_id']) + + last_station = self.stations.get(path[-1]) + if last_station and last_station['is_station']: + combined_path.append(path[-1]) + + return combined_path + + def _add_intra_system_connections(self) -> None: + """ + Adds connections between stations within the same solar system. + """ + stations_by_system: dict[int, list[int]] = {} + for station_id, station_data in self.stations.items(): + system_id = station_data['solar_system_id'] + stations_by_system.setdefault(system_id, []).append(station_id) + + for system_stations in stations_by_system.values(): + for i in range(len(system_stations)): + for j in range(i + 1, len(system_stations)): + self.graph.add_edge(system_stations[i], system_stations[j]) + + def _load_star_gate_connections(self) -> None: + """ + Loads stargate connections into the graph. + """ + connections = load_star_gate_connections() + for from_station, to_station in connections: + if from_station in self.stations and to_station in self.stations: + self.graph.add_edge(from_station, to_station) + + def _add_time_and_risk(self, ship: dict[str, Any]) -> None: + """ + Calculates and adds time and risk attributes to graph edges. + + Args: + ship: A dictionary containing ship parameters. + """ + for u, v in self.graph.edges(): + time = self._calculate_time(u, v, ship) + risk = self._calculate_risk(v, ship.get("gankers_areas", {})) + weight = time * ship.get("player_cost", 1) + ship.get("risk_cost", 1) * risk + self.graph[u][v]['time'] = time + self.graph[u][v]['risk'] = risk + self.graph[u][v]['weight'] = weight + + def _calculate_time(self, from_station: int, to_station: int, ship: dict[str, Any]) -> float: + """ + Calculates the time to travel between two stations. + + Args: + from_station: The starting station ID. + to_station: The destination station ID. + ship: A dictionary containing ship parameters. + + Returns: + The time in seconds to travel between the two stations. + """ + max_warp_speed = ship.get("max_warp_speed", 1) + max_subwarp_speed = ship.get("max_subwarp_speed", 1) + + # Inter-system travel (using stargates) + if self.stations[from_station]['solar_system_id'] != self.stations[to_station]['solar_system_id']: + return 9.0 # Fixed time for stargate jumps + + # Intra-system travel + position_from = self.stations[from_station]['position'] + position_to = self.stations[to_station]['position'] + warp_dist = sqrt(sum((position_from[i] - position_to[i]) ** 2 for i in range(3))) + + time_in_warp = self._calculate_time_in_warp(warp_dist, max_warp_speed, max_subwarp_speed) + time_in_warp += 2.0 # Add 2 seconds for initial delay + + return time_in_warp + + def _calculate_time_in_warp(self, warp_dist: float, max_warp_speed: float, max_subwarp_speed: float) -> float: + """ + Calculates the time spent in warp between two points. + + Args: + warp_dist: The distance to warp in meters. + max_warp_speed: The maximum warp speed of the ship in AU/s. + max_subwarp_speed: The maximum sub-warp speed of the ship in m/s. + + Returns: + The time in seconds spent in warp. + """ + AU_IN_M = 149_597_870_700 # Astronomical Unit in meters + + k_accel = max_warp_speed + k_decel = min(max_warp_speed / 3, 2) + warp_dropout_speed = min(max_subwarp_speed / 2, 100) + max_ms_warp_speed = max_warp_speed * AU_IN_M + + accel_dist = AU_IN_M + decel_dist = max_ms_warp_speed / k_decel + minimum_dist = accel_dist + decel_dist + + cruise_time = 0.0 + if minimum_dist > warp_dist: + max_ms_warp_speed = warp_dist * k_accel * k_decel / (k_accel + k_decel) + else: + cruise_time = (warp_dist - minimum_dist) / max_ms_warp_speed + + accel_time = log(max_ms_warp_speed / k_accel) / k_accel + decel_time = log(max_ms_warp_speed / warp_dropout_speed) / k_decel + + total_time = cruise_time + accel_time + decel_time + return total_time + + def _calculate_risk(self, to_station: int, gankers_areas: dict[str, float]) -> float: + """ + Calculates the risk of traveling to a station. + + Args: + to_station: The destination station ID. + gankers_areas: A dictionary mapping solar system names to risk factors. + + Returns: + A risk score as a float. + """ + station_data = self.stations[to_station] + security = station_data.get('security', 1.0) + solar_system_name = station_data.get('solar_system_name', '') + + # Base risk calculation based on security status + risk = 0.00004 + (1 - security) * 0.0004 + if security < 0.5: + risk += 0.005 + if security < 0.1: + risk += 0.5 + + risk += gankers_areas.get(solar_system_name, 0) + + return risk + + def _load_paths(self) -> dict[int, dict[int, float]]: + """ + Loads or generates the shortest path lengths between all pairs. + + Returns: + A dictionary of dictionaries containing path lengths. + """ + if PATHS_FILE.exists(): + with PATHS_FILE.open("rb") as f: + logger.info("Loading cached paths from file") + return pickle.load(f) + else: + logger.info("Generating path lengths and caching to file") + dist = dict(nx.all_pairs_dijkstra_path_length(self.graph)) + with PATHS_FILE.open("wb") as f: + pickle.dump(dist, f) + return dist diff --git a/backend/src/loader.py b/backend/src/loader.py new file mode 100644 index 0000000..7e536c1 --- /dev/null +++ b/backend/src/loader.py @@ -0,0 +1,186 @@ +import csv +import logging +from pathlib import Path +from typing import Any + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +BASE_DIR = Path(__file__).resolve().parent.parent / 'data' +MAP_DENORMALIZED_FILE = BASE_DIR / 'mapDenormalize.csv' +MAP_JUMPS_FILE = BASE_DIR / 'mapJumps.csv' +INV_TYPES_FILE = BASE_DIR / 'invTypes.csv' +REGION_IDS_FILE = BASE_DIR / 'mapRegions.csv' + +# Define constants +TYPE_ID_SOLAR_SYSTEM = 5 +GROUP_ID_STATION = 15 +GROUP_ID_STARGATE = 10 + +REGION_IDS = [ + 10000002, # The Forge + 10000016, # Lonetrek + 10000030, # Heimatar + 10000033, # The Citadel + 10000032, # Sinq Laison + 10000042, # Metropolis + 10000064, # Essence + 10000037, # Everyshore + 10000043, # Domain + 10000036, # Devoid + 10000052, # Kador + 10000067, # Genesis + 10000068, # Verge Vendor + 10000001, # Derelik + 10000020, # Tash-Murkon +] + +def load_map_data() -> tuple[dict[int, Any], dict[int, Any]]: + """ + Loads map data including stations and solar systems. + + Returns: + Tuple containing dictionaries of stations and solar systems. + """ + regions = read_regions() + stations: dict[int, Any] = {} + solar_systems: dict[int, Any] = {} + + with MAP_DENORMALIZED_FILE.open(newline='', encoding='utf-8') as csvfile: + reader = csv.reader(csvfile) + next(reader) # Skip header + for row in reader: + process_row(row, solar_systems, stations, regions) + + # Update station data with additional solar system and region info + for station_id, station_data in stations.items(): + ss_id = station_data['solar_system_id'] + solar_system = solar_systems.get(ss_id) + if solar_system: + station_data['solar_system_name'] = solar_system['item_name'] + station_data['solar_system_position'] = solar_system['position'] + station_data['region_id'] = solar_system['region_id'] + station_data['region_name'] = regions.get(station_data['region_id'], 'Unknown') + + return stations, solar_systems + +def process_row( + row: list[str], + solar_systems: dict[int, Any], + stations: dict[int, Any], + regions: dict[int, str] +) -> None: + """ + Processes a single row from the map denormalized CSV file. + + Args: + row: A list of CSV fields. + solar_systems: Dictionary to store solar system data. + stations: Dictionary to store station data. + regions: Dictionary of region IDs to region names. + """ + ( + item_id, type_id, group_id, solar_system_id, + _, region_id, _, x, y, z, _, + item_name, security, _, _ + ) = row + + try: + region_id_int = int(region_id) + except (ValueError, TypeError): + return + + if region_id_int not in REGION_IDS: + return + + item_id_int = int(item_id) + type_id_int = int(type_id) + group_id_int = int(group_id) + solar_system_id_int = int(solar_system_id) + + if type_id_int == TYPE_ID_SOLAR_SYSTEM: + solar_systems[item_id_int] = { + 'item_name': item_name, + 'position': (float(x), float(y), float(z)), + 'region_id': region_id_int, + 'region_name': regions.get(region_id_int, 'Unknown') + } + elif group_id_int in (GROUP_ID_STATION, GROUP_ID_STARGATE): + stations[item_id_int] = { + 'is_station': group_id_int == GROUP_ID_STATION, + 'item_name': item_name, + 'security': float(security), + 'position': (float(x), float(y), float(z)), + 'solar_system_id': solar_system_id_int, + 'region_id': region_id_int, + 'jumps': [] + } + +def load_items() -> dict[int, Any]: + """ + Loads item data from the inventory types CSV file. + + Returns: + Dictionary of items with type IDs as keys. + """ + items: dict[int, Any] = {} + with INV_TYPES_FILE.open(newline='', encoding='utf-8') as csvfile: + reader = csv.DictReader(csvfile) + for row in reader: + try: + type_id = int(row['typeID']) + items[type_id] = { + 'type_id': type_id, + 'group_id': int(row['groupID']), + 'type_name': row['typeName'], + 'description': row['description'], + 'mass': float(row['mass']) if row['mass'] else None, + 'volume': float(row['volume']) if row['volume'] else None, + 'capacity': float(row['capacity']) if row['capacity'] else None, + } + except ValueError as e: + logger.warning(f"Could not parse item with typeID {row.get('typeID')}: {e}") + except KeyError as e: + logger.error(f"Missing expected field in inventory types CSV: {e}") + return items + +def read_regions() -> dict[int, str]: + """ + Reads region data from the regions CSV file. + + Returns: + Dictionary mapping region IDs to region names. + """ + regions: dict[int, str] = {} + with REGION_IDS_FILE.open(newline='', encoding='utf-8') as csvfile: + reader = csv.DictReader(csvfile) + for row in reader: + try: + region_id = int(row['regionID']) + regions[region_id] = row['regionName'].strip() + except ValueError as e: + logger.warning(f"Could not parse region ID: {e}") + except KeyError as e: + logger.error(f"Missing expected field in regions CSV: {e}") + return regions + +def load_star_gate_connections() -> list[tuple[int, int]]: + """ + Loads stargate connections from the map jumps CSV file. + + Returns: + List of tuples representing stargate connections. + """ + star_gate_connections: list[tuple[int, int]] = [] + with MAP_JUMPS_FILE.open(newline='', encoding='utf-8') as csvfile: + reader = csv.DictReader(csvfile) + for row in reader: + try: + from_stargate = int(row['stargateID']) + to_stargate = int(row['destinationID']) + star_gate_connections.append((from_stargate, to_stargate)) + except ValueError as e: + logger.warning(f"Could not parse stargate connection: {e}") + except KeyError as e: + logger.error(f"Missing expected field in stargate jumps CSV: {e}") + return star_gate_connections diff --git a/backend/src/main.py b/backend/src/main.py new file mode 100644 index 0000000..c1a76f5 --- /dev/null +++ b/backend/src/main.py @@ -0,0 +1,94 @@ +import asyncio +import json +import logging +from typing import Any, Optional + +from fastapi import BackgroundTasks, FastAPI +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import StreamingResponse + +from manager import Manager +import esi.api as esi + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +app = FastAPI() +manager = Manager() + +async def start_manager() -> None: + """Periodically updates the manager in a separate thread to prevent blocking.""" + while True: + try: + logger.info("Starting manager update") + await asyncio.to_thread(manager.update) + logger.info("Manager update completed") + except Exception as error: + logger.exception("An error occurred during manager update: %s", error) + await asyncio.sleep(2) + +@app.on_event("startup") +async def startup_event() -> None: + """Initializes the manager update task on startup.""" + app.state.manager_task = asyncio.create_task(start_manager()) + +# Configure CORS +app.add_middleware( + CORSMiddleware, + allow_origins=["http://localhost:3000"], # Allows requests from specified origin + allow_credentials=True, + allow_methods=["*"], # Allows all HTTP methods + allow_headers=["*"], # Allows all HTTP headers +) + +@app.get("/") +async def read_root() -> dict: + """Root endpoint that returns a welcome message.""" + return {"Hello": "World"} + +@app.get("/items/{item_id}") +async def read_item(item_id: int, q: Optional[str] = None) -> dict: + """Retrieves an item by ID, optionally filtering by query parameter `q`.""" + return {"item_id": item_id, "q": q} + +@app.get("/route") +async def read_route() -> Any: + """Returns the current route from the manager.""" + return manager.route + +@app.get("/route_info") +async def read_route_info() -> Any: + """Returns route information from the manager.""" + return manager.route_info + +@app.post("/create_route") +async def create_route(background_tasks: BackgroundTasks) -> dict: + """Initiates route creation in the background.""" + background_tasks.add_task(manager.create_route) + return {"status": "Route creation initiated"} + +@app.get("/data_stream") +async def data_stream() -> StreamingResponse: + """Streams data updates to the client.""" + async def event_stream() -> Any: + while True: + data = { + "route": manager.route, + "route_info": manager.route_info, + "log": manager.log, + } + yield f"data: {json.dumps(data)}\n\n" + await asyncio.sleep(1) + return StreamingResponse(event_stream(), media_type="text/event-stream") + +@app.get("/logs") +async def get_logs() -> Any: + """Retrieves logs from the manager.""" + return manager.log_queue + +@app.post("/open_market_window/{type_id}") +async def open_market_window(type_id: int) -> Any: + """Opens a market window for a given `type_id` using the ESI API.""" + result = esi.open_market_window(type_id) + logger.info("Opened market window for type_id %s", type_id) + return result diff --git a/backend/src/manager.py b/backend/src/manager.py new file mode 100644 index 0000000..03df5d8 --- /dev/null +++ b/backend/src/manager.py @@ -0,0 +1,104 @@ +import threading + +from graph import Graph +from esi.api import get_location, set_waypoints, open_market_window +from market import Market +from arbitrage import arbitrage, snipe +from pathfinder import pathfinder +import route + + + +class Manager: + + def __init__(self): + self.ship = { + "location": get_location(), + "cargo": 600, + "max_warp_speed": 8.22, # AU/s + "max_subwarp_speed": 216.5, # m/s + "gankers_areas": {"Ahbazon": 0.5}, #["Miroitem", "Rancer", "Huola", "Ahbazon", "Uedama", "Sivala", "Tama", "Onatoh"], + "player_cost": 15_000_000 / 3600, # isk/second + "risk_cost": 60_000_000, # isk + "ship_cost": 50_000_000 # isk + } + self.graph = Graph(self.ship) + self.route = [] + self.route_info = {} + self.market = Market() + self.updating_orders = False + self.log = ["Test Log"] + + def route(self): + return self.route + + def route_info(self): + return self.route_info + + def get_log(self): + result = self.log + self.log = [] + return result + + def snipe_callback(self, updated_regions): + + if self.ship["location"] in self.graph.solar_systems: + current_region = self.graph.solar_systems[self.ship["location"]]["region_id"] + else: + current_region = self.graph.stations[self.ship["location"]]["region_id"] + + if current_region in updated_regions: + potential_snipe = snipe(self.market.orders, self.ship["cargo"], current_region) + if potential_snipe: + self.log.append("----------------") + self.log.append("Snipe: " + str(potential_snipe[1]) + " orders for " + str(potential_snipe[0]) + " ISK") + self.log.append("----------------") + + print("snipe possible", potential_snipe[0], "ISK", potential_snipe[1]) + + open_market_window(potential_snipe[1]["type_id"]) + + + def update(self): + self.log = self.log[-100:] + + location = get_location() + self.ship["location"] = location + + if len(self.route) == 1: + # Only the last step is left, so we are at the destination + self.log.append("Arrived at Destination") + self.route = [] + return + + for i, step in enumerate(self.route[:2]): + if i != 0 and step["location_id"] == location: + self.log.append("Arrived at " + step["location"]) + self.route = self.route[i:] + break + + if not self.updating_orders: + self.updating_orders = True + + def target_function(log): + try: + updated_regions = self.market.update_orders(log) + if updated_regions: + self.snipe_callback(updated_regions) + finally: + self.updating_orders = False + + update_thread = threading.Thread(target=target_function, args=(self.log,)) + update_thread.start() + + + def create_route(self): + order_matches = arbitrage(self.market.orders, self.ship["cargo"]) + order_matches = pathfinder(order_matches, self.graph, self.ship) + self.route, self.route_info = route.route(order_matches, self.graph, self.ship) + + set_waypoints([r["location_id"] for r in self.route[1:]]) + + print(self.route) + + \ No newline at end of file diff --git a/backend/src/market.py b/backend/src/market.py new file mode 100644 index 0000000..311b181 --- /dev/null +++ b/backend/src/market.py @@ -0,0 +1,203 @@ +import json +import logging +from dataclasses import dataclass, asdict +from datetime import datetime +from pathlib import Path +from typing import List, Dict + +from loader import load_items +from esi.api import get_region_orders + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +ORDERS_CACHE = Path(__file__).resolve().parent.parent / "cache" / "orders_snapshot_orders.json" + +REGION_IDS = [ + 10000002, # The Forge + 10000016, # Lonetrek + 10000030, # Heimatar + 10000033, # The Citadel + 10000032, # Sinq Laison + 10000042, # Metropolis + 10000064, # Essence + 10000037, # Everyshore + 10000043, # Domain + 10000036, # Devoid + 10000052, # Kador + 10000067, # Genesis + 10000068, # Verge Vendor + 10000001, # Derelik + 10000020, # Tash-Murkon +] + +@dataclass +class Order: + order_id: int + region_id: int + location_id: int + type_id: int + is_buy_order: bool + price: float + issued: datetime + volume_remain: int + item_name: str = '' + item_cargo_volume: float = 1.0 + +class Market: + """ + Manages market orders, including fetching, updating, and caching. + """ + + def __init__(self): + self.orders: List[Order] = [] + self.expires_times: Dict[int, datetime] = {} + + def get_orders(self) -> List[Order]: + """Retrieves the current list of market orders.""" + self.update_orders() + return self.orders + + def update_orders(self) -> List[int]: + """ + Updates market orders by fetching new data for expired regions. + + Returns: + A list of region IDs that were updated. + """ + updated_orders: List[Order] = [] + regions_updated: List[int] = [] + + for region_id in REGION_IDS: + if self.check_expired(region_id): + logger.info(f"Downloading orders for region {region_id}") + orders = self.download_orders(region_id) + updated_orders.extend(orders) + regions_updated.append(region_id) + else: + # Include existing orders for this region + updated_orders.extend( + [order for order in self.orders if order.region_id == region_id] + ) + + self.orders = updated_orders + return regions_updated + + def check_expired(self, region_id: int) -> bool: + """ + Checks if the cached orders for a region have expired. + + Args: + region_id: The ID of the region to check. + + Returns: + True if the orders have expired or are not cached, False otherwise. + """ + current_time = datetime.utcnow() + expiry_time = self.expires_times.get(region_id) + return expiry_time is None or current_time >= expiry_time + + def download_orders(self, region_id: int) -> List[Order]: + """ + Downloads orders for a given region and updates the expiry time. + + Args: + region_id: The ID of the region to download orders for. + + Returns: + A list of Order instances for the region. + """ + raw_orders, expiry = get_region_orders(region_id) + expiry_time = datetime.strptime(expiry, "%a, %d %b %Y %H:%M:%S %Z") + self.expires_times[region_id] = expiry_time + + orders = self.add_item_info_to_orders(raw_orders) + return orders + + def add_item_info_to_orders(self, raw_orders: List[Dict]) -> List[Order]: + """ + Enhances raw order data with item information and converts them into Order instances. + + Args: + raw_orders: A list of raw order dictionaries from the API. + + Returns: + A list of Order instances with additional item information. + """ + items = load_items() + orders: List[Order] = [] + + for raw_order in raw_orders: + item_info = items.get(raw_order['type_id'], {}) + order = Order( + order_id=raw_order['order_id'], + region_id=raw_order['region_id'], + location_id=raw_order['location_id'], + type_id=raw_order['type_id'], + is_buy_order=raw_order['is_buy_order'], + price=raw_order['price'], + issued=datetime.strptime(raw_order['issued'], "%Y-%m-%dT%H:%M:%S%z"), + volume_remain=raw_order['volume_remain'], + item_name=item_info.get('type_name', f"Not found: {raw_order['type_id']}"), + item_cargo_volume=item_info.get('volume', 1.0), + ) + orders.append(order) + + return orders + + def cache_orders(self) -> None: + """Caches the current list of orders to a JSON file.""" + try: + with ORDERS_CACHE.open("w", encoding="utf-8") as f: + # Serialize orders to dictionaries and handle datetime serialization + orders_data = [self._serialize_order(order) for order in self.orders] + json.dump(orders_data, f, ensure_ascii=False, indent=4) + logger.info("Orders cached successfully.") + except IOError as e: + logger.error(f"Error caching orders: {e}") + + def load_orders(self) -> List[Order]: + """ + Loads orders from the cache file. + + Returns: + A list of Order instances loaded from the cache. + """ + try: + with ORDERS_CACHE.open("r", encoding="utf-8") as f: + orders_data = json.load(f) + self.orders = [self._deserialize_order(data) for data in orders_data] + logger.info("Orders loaded from cache successfully.") + except (IOError, json.JSONDecodeError) as e: + logger.error(f"Error loading orders from cache: {e}") + self.orders = [] + + return self.orders + + def _serialize_order(self, order: Order) -> Dict: + """ + Serializes an Order instance to a dictionary suitable for JSON serialization. + + Args: + order: The Order instance to serialize. + + Returns: + A dictionary representation of the Order. + """ + order_dict = asdict(order) + # Convert datetime objects to ISO format strings + order_dict['issued'] = order.issued.isoformat() + return order_dict + + def _deserialize_order(self, data: Dict) -> Order: + """ + Deserializes a dictionary into an Order instance. + + Args: + data: The dictionary representation of an Order. + + Returns: + An Order instance. + """ + data['issued'] = datetime.fromisoformat(data['issued']) + return Order(**data) diff --git a/backend/src/pathfinder.py b/backend/src/pathfinder.py new file mode 100644 index 0000000..80e7a44 --- /dev/null +++ b/backend/src/pathfinder.py @@ -0,0 +1,145 @@ +import logging +from dataclasses import dataclass +from tqdm import tqdm + +from arbitrage import Trade +from graph import Graph + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +@dataclass +class Ship: + location: int + cargo: int + max_warp_speed: float + max_subwarp_speed: float + gankers_areas: list[str] + player_cost: float + risk_cost: float + + +def pathfinder(order_matches: list[Trade], graph: Graph, ship: Ship) -> list[Trade]: + """ + Processes order matches by filtering and calculating net profit. + + Args: + order_matches: List of trade opportunities. + graph: Navigation graph. + ship: Ship parameters. + + Returns: + A list of filtered and annotated trade opportunities. + """ + order_matches = filter_order_matches_not_in_graph(order_matches, graph) + logger.info("After filtering order matches not in graph: %d", len(order_matches)) + + order_matches = filter_order_matches_same_region(order_matches, graph) + logger.info("After filtering order matches to same region: %d", len(order_matches)) + + order_matches = calculate_net_profit(order_matches, graph, ship) + + order_matches = filter_order_matches_low_profit(order_matches) + logger.info("After filtering low profit order matches: %d", len(order_matches)) + + return order_matches + +def filter_order_matches_not_in_graph(order_matches: list[Trade], graph: Graph) -> list[Trade]: + """ + Filters out order matches where either the from_station or to_station is not in the graph. + + Args: + order_matches: List of trade opportunities. + graph: Navigation graph. + + Returns: + Filtered list of trade opportunities. + """ + return [ + order_match for order_match in order_matches + if order_match.from_station in graph.graph.nodes and order_match.to_station in graph.graph.nodes + ] + +def filter_order_matches_same_region(order_matches: list[Trade], graph: Graph) -> list[Trade]: + """ + Filters out order matches where the from_region and to_region are different. + Keeps only order matches within the same region. + + Args: + order_matches: List of trade opportunities. + graph: Navigation graph. + + Returns: + Filtered list of trade opportunities where both stations are in the same region. + """ + filtered_order_matches = [] + + for order_match in order_matches: + from_station = order_match.from_station + to_station = order_match.to_station + + from_station_data = graph.stations.get(from_station) + to_station_data = graph.stations.get(to_station) + + if not from_station_data or not to_station_data: + continue # Skip if station data is missing + + from_solar_system_id = from_station_data["solar_system_id"] + to_solar_system_id = to_station_data["solar_system_id"] + + from_region = graph.solar_systems[from_solar_system_id]["region_id"] + to_region = graph.solar_systems[to_solar_system_id]["region_id"] + + if from_region == to_region: + filtered_order_matches.append(order_match) + + return filtered_order_matches + +def calculate_net_profit(order_matches: list[Trade], graph: Graph, ship: Ship) -> list[Trade]: + """ + Calculates the net profit for each order match by subtracting transport cost from gross profit. + + Args: + order_matches: List of trade opportunities. + graph: Navigation graph. + ship: Ship parameters. + + Returns: + List of trade opportunities with net_profit calculated. + """ + logger.info("Calculating estimated profits") + for order_match in tqdm(order_matches): + try: + # Calculate transport cost from ship location to from_station + cost_to_source = graph.shortest_path_length(ship.location, order_match.from_station) + # Calculate transport cost from from_station to to_station + cost_to_destination = graph.shortest_path_length(order_match.from_station, order_match.to_station) + + # Check if paths exist + if cost_to_source is None or cost_to_destination is None: + logger.warning(f"No path found for order match: {order_match}") + order_match.net_profit = None + continue + + total_transport_cost = cost_to_source + cost_to_destination + order_match.net_profit = order_match.gross_profit - total_transport_cost + except Exception as e: + logger.error(f"Error calculating net profit for order match {order_match}: {e}") + order_match.net_profit = None + + return order_matches + +def filter_order_matches_low_profit(order_matches: list[Trade]) -> list[Trade]: + """ + Filters out order matches with non-positive net profit. + + Args: + order_matches: List of trade opportunities with net_profit calculated. + + Returns: + Filtered list of profitable trade opportunities. + """ + return [ + order_match for order_match in order_matches + if order_match.net_profit and order_match.net_profit > 0 + ] diff --git a/backend/src/route.py b/backend/src/route.py new file mode 100644 index 0000000..ec099b1 --- /dev/null +++ b/backend/src/route.py @@ -0,0 +1,208 @@ +import logging +from dataclasses import dataclass, field +from typing import Optional + +from tqdm import tqdm + +from graph import Graph +from pathfinder import Ship +from arbitrage import Trade + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +MAX_TRADES_TO_CONSIDER = 20_000 +MAX_CAPITAL = 100_000_000 + +@dataclass +class RouteStep: + location_id: int + station_name: str + actions: list[dict] = field(default_factory=list) + +Route = list[RouteStep] + +def route(trades: list[Trade], graph: Graph, ship: Ship) -> tuple[Optional[Route], Optional[dict]]: + """ + Given a list of trades, find the best route starting from the ship's location. + + Args: + trades: List of Trade instances. + graph: Navigation graph. + ship: Ship parameters. + + Returns: + A tuple containing the best route and route information. + """ + trades.sort( + key=lambda trade: trade.gross_profit / (trade.from_price * trade.quantity) if trade.from_price * trade.quantity > 0 else 0, + reverse=True + ) + + best_profit_rate = 0.0 + best_route: Optional[Route] = None + best_route_info: Optional[dict] = None + + seen_routes = set() + + for trade in tqdm(trades[:MAX_TRADES_TO_CONSIDER], desc="Evaluating trades"): + route_stations = (ship.location, trade.from_station, trade.to_station) + + if route_stations in seen_routes: + continue + seen_routes.add(route_stations) + + optimized_trades = select_trades(list(route_stations), trades, ship) + total_profit = sum(t.gross_profit for t in optimized_trades) + + if not optimized_trades or total_profit <= 0: + continue + + try: + path = graph.shortest_path(ship.location, trade.from_station)[:-1] + path += graph.shortest_path(trade.from_station, trade.to_station) + except Exception as e: + logger.error(f"Error finding path: {e}") + continue + + risk = sum(graph.graph[u][v]['risk'] for u, v in zip(path, path[1:])) + transport_time = sum(graph.graph[u][v]['time'] for u, v in zip(path, path[1:])) + + capital = getattr(ship, 'ship_cost', 0.0) + sum(t.from_price * t.quantity for t in optimized_trades) + + if transport_time == 0: + logger.warning("Transport time is zero, skipping this route to avoid division by zero.") + continue + + net_profit = total_profit - risk * capital + profit_rate = net_profit / transport_time + + if profit_rate > best_profit_rate: + logger.info( + f"New best profit rate: {profit_rate * 3600:.2f} per hour, " + f"Risk: {risk:.4f}, Capital: {capital:.2f}, " + f"Transport time: {transport_time:.2f} sec, Gross profit: {total_profit:.2f}" + ) + formatted_route = graph.formatted_route(path) + final_route = set_actions(formatted_route, optimized_trades) + best_route = final_route + best_route_info = { + "profit_rate": profit_rate, + "risk": risk, + "capital": capital, + "transport_time": transport_time, + "gross_profit": total_profit, + "net_profit": net_profit, + } + best_profit_rate = profit_rate + + return best_route, best_route_info + +def select_trades(stations: list[int], trades: list[Trade], ship: Ship) -> list[Trade]: + """ + Given a list of stations, find the optimal trades to add. + + Args: + stations: List of station IDs. + trades: List of Trade instances. + ship: Ship parameters. + + Returns: + A list of optimized Trade instances. + """ + filtered_trades = [ + trade for trade in trades + if trade.from_station in stations and trade.to_station in stations and + stations.index(trade.from_station) < stations.index(trade.to_station) + ] + + filtered_trades.sort( + key=lambda trade: trade.gross_profit / trade.cargo if trade.cargo > 0 else 0, + reverse=True + ) + + used_orders = {} + optimized_trades = [] + cargo = 0.0 + capital = 0.0 + + for trade in filtered_trades: + quantity = trade.quantity + from_order, to_order = trade.from_order_id, trade.to_order_id + + from_remaining = used_orders.get(from_order, quantity) + to_remaining = used_orders.get(to_order, quantity) + + max_quantity_by_cargo = int((ship.cargo - cargo) // trade.item_cargo_volume) + max_quantity_by_capital = int((MAX_CAPITAL - capital) // trade.from_price) + + max_quantity = min(quantity, from_remaining, to_remaining, max_quantity_by_cargo, max_quantity_by_capital) + + if max_quantity <= 0: + continue + + cargo += trade.item_cargo_volume * max_quantity + capital += trade.from_price * max_quantity + used_orders[from_order] = from_remaining - max_quantity + used_orders[to_order] = to_remaining - max_quantity + + # Create a new Trade instance with updated quantities + updated_trade = Trade( + from_station=trade.from_station, + to_station=trade.to_station, + from_order_id=trade.from_order_id, + to_order_id=trade.to_order_id, + item_name=trade.item_name, + type_id=trade.type_id, + item_cargo_volume=trade.item_cargo_volume, + from_price=trade.from_price, + to_price=trade.to_price, + quantity=max_quantity, + cargo=trade.item_cargo_volume * max_quantity, + gross_profit=(trade.to_price - trade.from_price) * max_quantity, + net_profit=None # Will be calculated later if needed + ) + + optimized_trades.append(updated_trade) + + if cargo >= ship.cargo or capital >= MAX_CAPITAL: + break + + return optimized_trades + +def set_actions(route: Route, trades: list[Trade]) -> Route: + """ + Sets the actions (buy/sell) at each step in the route based on the trades. + + Args: + route: The route steps. + trades: List of Trade instances. + + Returns: + The route with actions set. + """ + bought_items = set() + for trade in trades: + for step in route: + if trade.from_station == step.location_id: + step.actions.append({ + "action_type": "buy", + "item": trade.item_name, + "type_id": trade.type_id, + "quantity": trade.quantity, + "price": trade.from_price, + }) + bought_items.add(trade.type_id) + if trade.to_station == step.location_id and trade.type_id in bought_items: + step.actions.append({ + "action_type": "sell", + "item": trade.item_name, + "type_id": trade.type_id, + "quantity": trade.quantity, + "price": trade.to_price, + }) + + for step in route: + step.actions.sort(key=lambda action: (action["action_type"], action["item"], action["price"])) + + return route -- cgit v1.3