Created
September 23, 2024 17:47
-
-
Save jamesaphoenix/170bf045225d23831739a2d430045a14 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import os | |
| import uuid | |
| from typing import List | |
| import requests | |
| from fastapi import HTTPException | |
| def get_profiles(api_url: str, headers: dict) -> List[dict]: | |
| """ | |
| Get the profiles associated with the account. | |
| """ | |
| try: | |
| response = requests.get(f"{api_url}/v2/profiles", headers=headers, timeout=10) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.RequestException as e: | |
| raise HTTPException( | |
| status_code=500, detail=f"Error retrieving profiles: {e}" | |
| ) from e | |
| def find_profile(profiles: List[dict], company_name: str): | |
| """ | |
| Find the profile ID based on the company name. | |
| """ | |
| for profile in profiles: | |
| if profile["fullName"] == company_name: | |
| return profile["id"] | |
| raise HTTPException(detail="Profile not found", status_code=404) | |
| def get_balances(api_url: str, profile_id: str, headers: dict, balance_type="STANDARD"): | |
| """ | |
| Get the balances associated with the profile. | |
| """ | |
| try: | |
| response = requests.get( | |
| f"{api_url}/v4/profiles/{profile_id}/balances?types={balance_type}", | |
| headers=headers, | |
| timeout=15, | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.RequestException as e: | |
| raise HTTPException( | |
| status_code=500, detail=f"Error retrieving balances: {e}" | |
| ) from e | |
| def find_balance_id(balances: List[dict], currency: str, balance_type: str): | |
| """Find the balance ID based on currency and type.""" | |
| for balance in balances: | |
| if balance["currency"] == currency and balance["type"] == balance_type: | |
| return balance["id"] | |
| raise HTTPException(detail="Balance ID not found", status_code=404) | |
| def move_money_between_balances( | |
| api_url: str, | |
| headers: dict, | |
| profile_id: str, | |
| source_balance_id: int, | |
| target_balance_id: int, | |
| amount, | |
| ): | |
| """ | |
| Move money between balances. | |
| """ | |
| data = { | |
| "sourceBalanceId": source_balance_id, | |
| "targetBalanceId": target_balance_id, | |
| "amount": {"value": amount, "currency": "GBP"}, | |
| } | |
| headers.update({"X-idempotence-uuid": str(uuid.uuid4())}) | |
| try: | |
| response = requests.post( | |
| f"{api_url}/v2/profiles/{profile_id}/balance-movements", | |
| headers=headers, | |
| json=data, | |
| timeout=30, | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |
| except requests.RequestException as e: | |
| raise HTTPException( | |
| status_code=500, detail=f"Error moving money between balances: {e}" | |
| ) from e | |
| def calculate_investment_strategy( | |
| balances: List[dict], | |
| daily_investment: float, | |
| target_percentage: float, | |
| minimum_uk_cash_balance: float, | |
| ): | |
| """ | |
| Calculate the investment strategy based on the current balances. | |
| """ | |
| # Sum the total worth of both balances and stock balances | |
| total_worth = sum( | |
| balance["totalWorth"]["value"] | |
| for balance in balances | |
| if balance["currency"] == "GBP" and balance["type"] == "STANDARD" | |
| ) | |
| total_stock_worth = sum( | |
| balance["totalWorth"]["value"] | |
| for balance in balances | |
| if balance["currency"] == "GBP" and balance["type"] == "STANDARD" | |
| ) | |
| combined_worth = total_worth + total_stock_worth | |
| target_investment = combined_worth * target_percentage / 100 | |
| # Current invested amount in the stock balance | |
| current_investment = sum( | |
| stock["totalWorth"]["value"] | |
| for stock in balances | |
| if stock["currency"] == "GBP" and stock["type"] == "SAVINGS" | |
| ) | |
| # Calculate the difference from the target investment | |
| remaining_investment_needed = target_investment - current_investment | |
| # Check if today's investment should proceed | |
| if ( | |
| remaining_investment_needed > daily_investment | |
| and total_worth > minimum_uk_cash_balance | |
| ): | |
| print(f"Today's investment: £{daily_investment}") | |
| return daily_investment | |
| else: | |
| print("No investment needed today") | |
| return 0 | |
| def invest_in_stocks( | |
| target_percentage: float, | |
| minimum_uk_cash_balance: float, | |
| daily_investment: float, | |
| company_name="Just Understanding Data Ltd", | |
| ): | |
| """ | |
| Invest in stocks based on the investment strategy. | |
| """ | |
| api_url = "https://api.transferwise.com" | |
| headers = {"Authorization": "Bearer " + os.environ["WISE_API_KEY"]} | |
| # Main Execution: | |
| profiles = get_profiles(api_url, headers) | |
| profile_id = find_profile(profiles, company_name) | |
| if profile_id is None: | |
| raise HTTPException(detail="Profile ID not found", status_code=404) | |
| balances = get_balances(api_url, profile_id, headers, "STANDARD,SAVINGS") | |
| amount_to_invest = calculate_investment_strategy( | |
| balances, | |
| daily_investment=daily_investment, | |
| target_percentage=target_percentage, | |
| minimum_uk_cash_balance=minimum_uk_cash_balance, | |
| ) | |
| if amount_to_invest > 0: | |
| source_balance_id = find_balance_id(balances, "GBP", "STANDARD") | |
| target_balance_id = find_balance_id(balances, "GBP", "SAVINGS") | |
| if not source_balance_id or not target_balance_id: | |
| raise HTTPException( | |
| detail="source_balance_id or target_balance_id are not valid", | |
| status_code=404, | |
| ) | |
| result = move_money_between_balances( | |
| api_url, | |
| headers, | |
| profile_id, | |
| source_balance_id, | |
| target_balance_id, | |
| amount_to_invest, | |
| ) | |
| print(f"Invested £{amount_to_invest} in stocks") | |
| return result | |
| else: | |
| print("No investment needed today") | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment