update

.chainlit/config.toml

@@ -0,0 +1,127 @@
+[project]
+# Whether to enable telemetry (default: true). No personal data is collected.
+enable_telemetry = true
+
+
+# List of environment variables to be provided by each user to use the app.
+user_env = []
+
+# Duration (in seconds) during which the session is saved when the connection is lost
+session_timeout = 3600
+
+# Enable third parties caching (e.g LangChain cache)
+cache = false
+
+# Authorized origins
+allow_origins = ["*"]
+
+# Follow symlink for asset mount (see https://github.com/Chainlit/chainlit/issues/317)
+# follow_symlink = false
+
+[features]
+# Show the prompt playground
+prompt_playground = true
+
+# Process and display HTML in messages. This can be a security risk (see https://stackoverflow.com/questions/19603097/why-is-it-dangerous-to-render-user-generated-html-or-javascript)
+unsafe_allow_html = false
+
+# Process and display mathematical expressions. This can clash with "$" characters in messages.
+latex = false
+
+# Automatically tag threads with the current chat profile (if a chat profile is used)
+auto_tag_thread = true
+
+# Authorize users to spontaneously upload files with messages
+[features.spontaneous_file_upload]
+    enabled = true
+    accept = ["*/*"]
+    max_files = 20
+    max_size_mb = 500
+
+[features.audio]
+    # Threshold for audio recording
+    min_decibels = -45
+    # Delay for the user to start speaking in MS
+    initial_silence_timeout = 3000
+    # Delay for the user to continue speaking in MS. If the user stops speaking for this duration, the recording will stop.
+    silence_timeout = 1500
+    # Above this duration (MS), the recording will forcefully stop.
+    max_duration = 15000
+    # Duration of the audio chunks in MS
+    chunk_duration = 1000
+    # Sample rate of the audio
+    sample_rate = 44100
+
+[UI]
+# Name of the app and chatbot.
+name = "Giino"
+
+# Show the readme while the thread is empty.
+show_readme_as_default = true
+
+# Description of the app and chatbot. This is used for HTML tags.
+# description = "AI-powered unification and execution protocol assitant for web3"
+
+# Large size content are by default collapsed for a cleaner ui
+default_collapse_content = true
+
+# The default value for the expand messages settings.
+default_expand_messages = false
+
+# Hide the chain of thought details from the user in the UI.
+hide_cot = false
+
+# Link to your github repo. This will add a github button in the UI's header.
+# github = ""
+
+# Specify a CSS file that can be used to customize the user interface.
+# The CSS file can be served from the public directory or via an external link.
+# custom_css = "/public/test.css"
+custom_css = '/public/stylesheet.css'
+
+# Specify a Javascript file that can be used to customize the user interface.
+# The Javascript file can be served from the public directory.
+# custom_js = "/public/test.js"
+custom_js = "/public/script.js"
+
+# Specify a custom font url.
+# custom_font = "https://fonts.googleapis.com/css2?family=Inter:wght@400;500;700&display=swap"
+
+# Specify a custom meta image url.
+# custom_meta_image_url = "https://chainlit-cloud.s3.eu-west-3.amazonaws.com/logo/chainlit_banner.png"
+
+# Specify a custom build directory for the frontend.
+# This can be used to customize the frontend code.
+# Be careful: If this is a relative path, it should not start with a slash.
+# custom_build = "./public/build"
+
+[UI.theme]
+    layout = "wide"
+    font_family = "Inter, sans-serif"
+# Override default MUI light theme. (Check theme.ts)
+[UI.theme.light]
+#    background = "#FFFFFF"
+    paper = "#FFFFFF"
+
+    [UI.theme.light.primary]
+        main = "#007BFF"
+        dark = "#1C1C1C"
+        light = "#FAFAFA"
+
+# Override default MUI dark theme. (Check theme.ts)
+[UI.theme.dark]
+    background = "#1C1C1C"
+    paper = "#202020"
+
+    [UI.theme.dark.primary]
+        main = "#007BFF"
+        dark = "#1C1C1C"
+        light = "#202020"
+
+
+#Button Color: #1C4DF0
+#Button Color 2: #062C8A
+
+
+[meta]
+generated_by = "1.1.202"

.chainlit/translations/en-US.json

@@ -0,0 +1,231 @@
+{
+    "components": {
+        "atoms": {
+            "buttons": {
+                "userButton": {
+                    "menu": {
+                        "settings": "Settings",
+                        "settingsKey": "S",
+                        "APIKeys": "API Keys",
+                        "logout": "Logout"
+                    }
+                }
+            }
+        },
+        "molecules": {
+            "newChatButton": {
+                "newChat": "New Chat"
+            },
+            "tasklist": {
+                "TaskList": {
+                    "title": "\ud83d\uddd2\ufe0f Task List",
+                    "loading": "Loading...",
+                    "error": "An error occured"
+                }
+            },
+            "attachments": {
+                "cancelUpload": "Cancel upload",
+                "removeAttachment": "Remove attachment"
+            },
+            "newChatDialog": {
+                "createNewChat": "Create new chat?",
+                "clearChat": "This will clear the current messages and start a new chat.",
+                "cancel": "Cancel",
+                "confirm": "Confirm"
+            },
+            "settingsModal": {
+                "settings": "Settings",
+                "expandMessages": "Expand Messages",
+                "hideChainOfThought": "Hide Chain of Thought",
+                "darkMode": "Dark Mode"
+            },
+            "detailsButton": {
+                "using": "Using",
+                "running": "Running",
+                "took_one": "Took {{count}} step",
+                "took_other": "Took {{count}} steps"
+            },
+            "auth": {
+                "authLogin": {
+                    "title": "Login to access the app.",
+                    "form": {
+                        "email": "Email address",
+                        "password": "Password",
+                        "noAccount": "Don't have an account?",
+                        "alreadyHaveAccount": "Already have an account?",
+                        "signup": "Sign Up",
+                        "signin": "Sign In",
+                        "or": "OR",
+                        "continue": "Continue",
+                        "forgotPassword": "Forgot password?",
+                        "passwordMustContain": "Your password must contain:",
+                        "emailRequired": "email is a required field",
+                        "passwordRequired": "password is a required field"
+                    },
+                    "error": {
+                        "default": "Unable to sign in.",
+                        "signin": "Try signing in with a different account.",
+                        "oauthsignin": "Try signing in with a different account.",
+                        "redirect_uri_mismatch": "The redirect URI is not matching the oauth app configuration.",
+                        "oauthcallbackerror": "Try signing in with a different account.",
+                        "oauthcreateaccount": "Try signing in with a different account.",
+                        "emailcreateaccount": "Try signing in with a different account.",
+                        "callback": "Try signing in with a different account.",
+                        "oauthaccountnotlinked": "To confirm your identity, sign in with the same account you used originally.",
+                        "emailsignin": "The e-mail could not be sent.",
+                        "emailverify": "Please verify your email, a new email has been sent.",
+                        "credentialssignin": "Sign in failed. Check the details you provided are correct.",
+                        "sessionrequired": "Please sign in to access this page."
+                    }
+                },
+                "authVerifyEmail": {
+                    "almostThere": "You're almost there! We've sent an email to ",
+                    "verifyEmailLink": "Please click on the link in that email to complete your signup.",
+                    "didNotReceive": "Can't find the email?",
+                    "resendEmail": "Resend email",
+                    "goBack": "Go Back",
+                    "emailSent": "Email sent successfully.",
+                    "verifyEmail": "Verify your email address"
+                },
+                "providerButton": {
+                    "continue": "Continue with {{provider}}",
+                    "signup": "Sign up with {{provider}}"
+                },
+                "authResetPassword": {
+                    "newPasswordRequired": "New password is a required field",
+                    "passwordsMustMatch": "Passwords must match",
+                    "confirmPasswordRequired": "Confirm password is a required field",
+                    "newPassword": "New password",
+                    "confirmPassword": "Confirm password",
+                    "resetPassword": "Reset Password"
+                },
+                "authForgotPassword": {
+                    "email": "Email address",
+                    "emailRequired": "email is a required field",
+                    "emailSent": "Please check the email address {{email}} for instructions to reset your password.",
+                    "enterEmail": "Enter your email address and we will send you instructions to reset your password.",
+                    "resendEmail": "Resend email",
+                    "continue": "Continue",
+                    "goBack": "Go Back"
+                }
+            }
+        },
+        "organisms": {
+            "chat": {
+                "history": {
+                    "index": {
+                        "showHistory": "Show history",
+                        "lastInputs": "Last Inputs",
+                        "noInputs": "Such empty...",
+                        "loading": "Loading..."
+                    }
+                },
+                "inputBox": {
+                    "input": {
+                        "placeholder": "Type your message here..."
+                    },
+                    "speechButton": {
+                        "start": "Start recording",
+                        "stop": "Stop recording"
+                    },
+                    "SubmitButton": {
+                        "sendMessage": "Send message",
+                        "stopTask": "Stop Task"
+                    },
+                    "UploadButton": {
+                        "attachFiles": "Attach files"
+                    },
+                    "waterMark": {
+                        "text": ""
+                    }
+                },
+                "Messages": {
+                    "index": {
+                        "running": "Running",
+                        "executedSuccessfully": "executed successfully",
+                        "failed": "failed",
+                        "feedbackUpdated": "Feedback updated",
+                        "updating": "Updating"
+                    }
+                },
+                "dropScreen": {
+                    "dropYourFilesHere": "Drop your files here"
+                },
+                "index": {
+                    "failedToUpload": "Failed to upload",
+                    "cancelledUploadOf": "Cancelled upload of",
+                    "couldNotReachServer": "Could not reach the server",
+                    "continuingChat": "Continuing previous chat"
+                },
+                "settings": {
+                    "settingsPanel": "Settings panel",
+                    "reset": "Reset",
+                    "cancel": "Cancel",
+                    "confirm": "Confirm"
+                }
+            },
+            "threadHistory": {
+                "sidebar": {
+                    "filters": {
+                        "FeedbackSelect": {
+                            "feedbackAll": "Feedback: All",
+                            "feedbackPositive": "Feedback: Positive",
+                            "feedbackNegative": "Feedback: Negative"
+                        },
+                        "SearchBar": {
+                            "search": "Search"
+                        }
+                    },
+                    "DeleteThreadButton": {
+                        "confirmMessage": "This will delete the thread as well as it's messages and elements.",
+                        "cancel": "Cancel",
+                        "confirm": "Confirm",
+                        "deletingChat": "Deleting chat",
+                        "chatDeleted": "Chat deleted"
+                    },
+                    "index": {
+                        "pastChats": "Past Chats"
+                    },
+                    "ThreadList": {
+                        "empty": "Empty...",
+                        "today": "Today",
+                        "yesterday": "Yesterday",
+                        "previous7days": "Previous 7 days",
+                        "previous30days": "Previous 30 days"
+                    },
+                    "TriggerButton": {
+                        "closeSidebar": "Close sidebar",
+                        "openSidebar": "Open sidebar"
+                    }
+                },
+                "Thread": {
+                    "backToChat": "Go back to chat",
+                    "chatCreatedOn": "This chat was created on"
+                }
+            },
+            "header": {
+                "chat": "Chat",
+                "readme": "Readme"
+            }
+        }
+    },
+    "hooks": {
+        "useLLMProviders": {
+            "failedToFetchProviders": "Failed to fetch providers:"
+        }
+    },
+    "pages": {
+        "Design": {},
+        "Env": {
+            "savedSuccessfully": "Saved successfully",
+            "requiredApiKeys": "Required API Keys",
+            "requiredApiKeysInfo": "To use this app, the following API keys are required. The keys are stored on your device's local storage."
+        },
+        "Page": {
+            "notPartOfProject": "You are not part of this project."
+        },
+        "ResumeButton": {
+            "resumeChat": "Resume Chat"
+        }
+    }
+}

public/script.js

@@ -0,0 +1,113 @@
+document.addEventListener('DOMContentLoaded', () => {
+    const items = { ...localStorage };
+    const isAuthed = items?.token;
+
+    const wdgt = {
+        idBox: 'wdgt',
+        url_widget: 'https://next-starter.thirdweb-example.com/',
+        url_style: 'http://localhost:8080/css/widget.css',
+
+        init: function (identifier, byClass = false) {
+            console.log("Begin Widget initialization");
+
+            const observer = new MutationObserver((mutations, observer) => {
+                let elements = [];
+                if (byClass) {
+                    elements = document.querySelectorAll(`.${identifier}`);
+                    console.log(`Looking for elements by class: .${identifier}`);
+                } else {
+                    const element = document.getElementById(identifier || this.idBox);
+                    if (element) {
+                        elements.push(element);
+                    }
+                    console.log(`Looking for element by ID: #${identifier || this.idBox}`);
+                }
+
+                console.log(`Found elements:`, elements);
+
+                if (elements.length > 0 && elements[0] != null) {
+                    observer.disconnect(); // Stop observing once elements are found
+                    try {
+                        const XHR = ("onload" in new XMLHttpRequest()) ? XMLHttpRequest : XDomainRequest;
+                        const xhr = new XHR();
+                        xhr.open('GET', this.url_widget, true);
+
+                        xhr.onload = function () {
+                            if (this.status >= 200 && this.status < 300) {
+                                elements.forEach(element => {
+                                    element.innerHTML = this.responseText;
+                                });
+                            } else {
+                                console.error('Failed to load widget content. Status:', this.status);
+                            }
+                        };
+
+                        xhr.onerror = function () {
+                            console.error('onerror', this.status);
+                        };
+
+                        xhr.send();
+                    } catch (error) {
+                        console.error('Error loading widget:', error);
+                    }
+                }
+            });
+
+            // Start observing the document body for changes
+            observer.observe(document.body, { childList: true, subtree: true });
+
+            // Initial check in case the elements are already present
+            let elements = [];
+            if (byClass) {
+                elements = document.querySelectorAll(`.${identifier}`);
+            } else {
+                const element = document.getElementById(identifier || this.idBox);
+                if (element) {
+                    elements.push(element);
+                }
+            }
+
+            if (elements.length > 0 && elements[0] != null) {
+                observer.disconnect(); // Stop observing if elements are found initially
+                try {
+                    const XHR = ("onload" in new XMLHttpRequest()) ? XMLHttpRequest : XDomainRequest;
+                    const xhr = new XHR();
+                    xhr.open('GET', this.url_widget, true);
+
+                    xhr.onload = function () {
+                        if (this.status >= 200 && this.status < 300) {
+                            elements.forEach(element => {
+                                element.innerHTML = this.responseText;
+                            });
+                        } else {
+                            console.error('Failed to load widget content. Status:', this.status);
+                        }
+                    };
+
+                    xhr.onerror = function () {
+                        console.error('onerror', this.status);
+                    };
+
+                    xhr.send();
+                } catch (error) {
+                    console.error('Error loading widget:', error);
+                }
+            }
+        },
+
+        addStyle: function () {
+            const style = document.createElement('link');
+            style.rel = 'stylesheet';
+            style.type = 'text/css';
+            style.href = this.url_style;
+            document.head.appendChild(style);
+        }
+    };
+
+    if (isAuthed) {
+//        alert('You are authed');
+    } else {
+        // alert('You are not authed');
+        // wdgt.init('root', false); // Change 'true' to 'false' if you want to use an ID instead of class
+    }
+});

public/stylesheet.css

@@ -0,0 +1,3 @@
+.watermark, .css-1705j0v, .css-1rplq84 {
+    display: none!important
+}

src/config/nemoguardrails/config.yml

@@ -0,0 +1,27 @@
+# config.yml
+models:
+  - type: main
+    engine: openai
+    model: gpt-3.5-turbo-instruct
+
+rails:
+  # Input rails are invoked when new input from the user is received.
+  input:
+    flows:
+      - check jailbreak
+      - mask sensitive data on input
+
+  # Output rails are triggered after a bot message has been generated.
+  output:
+    flows:
+      - self check facts
+      - self check hallucination
+      - activefence moderation
+
+  config:
+    # Configure the types of entities that should be masked on user input.
+    sensitive_data_detection:
+      input:
+        entities:
+          - PERSON
+          - EMAIL_ADDRESS

src/config/telzho-430515-c2622883c30e.json

@@ -0,0 +1,13 @@
+{
+  "type": "service_account",
+  "project_id": "telzho-430515",
+  "private_key_id": "c2622883c30ecd05999180754516650bf6b692c5",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCa/PmlpE9CQVqb\nwsJHyDE8PCaieLWlPvIgslDArY36nvvtIaZuVkMjQkUyj4qJow/8sLF4PWt6ROMC\n+y5tZoPVk917EXKTCLySU/Q0lhZ3Y7Ftq+Y1Rilw4Mea0SRm0oFNEnnhmhBW3TCX\nvoFd+1BaEpK+lP7/4gN6FIjNZkOmqis0ikAlMppt0/Uni3HXozfY8o4Zeao41ebu\nJudYKfV2q9aizM+ZP615Dy225H1bQWEUKY9spvFt+HFat73mIFzFf6caFh04fVcf\nDdslEfJ9DAQESiI0QTLITyNmMxyl7h0E7Wc9saeCmQ2B8lc7L/HaonotveC7gdx1\nrEKi3LnnAgMBAAECggEAFzKzkIjIxDtRUJcUyqG68i/JU0WLYAcik/pHU/pQ51On\nPusWeiq3NIPVelKPLeLA9dWMpVlWyPTz9ZzwagzkwVxh05Xc9we2Z0CMVi1Q14s4\nd23ksHymn0VFxcp1ZF6Fs/+GoVnqBYETUOMsLIun5JN7G245whsWN3jzW6sUMnxy\nMRCMxl2SMkLvdjyeAOwUSaEj+c9/aSHW9mu90RUhZDxKY9xIj7ldQKC+xuVkYHBV\nwz2TQr0sIZz4aC/T+e6c6jPqVP8sWkA/QU1QhcpySzDe4ih3xbG4Hve/ZkO+h3KE\nEcw+BwDBtHLWJV2zlmvGX84Vz7VS1BFlBRZQWrFIIQKBgQDW09iZhNffer3wo04M\nRG1C5idRGL7JWWudoJbenRcYR8hco3A0WHTSicS5AV6QdN5eO8f4HQaI59TMtFWm\nGbun+d5uNRg6egIO4Yr9Am3xIpSzvulrA0d8V6+aSWkP/hCf+w0tB/AJKeIVxTgr\nCO9L5n1Wnf3+az1dL4rIi1Z4UwKBgQC4sTY2pNEYI8WgeuCfTCvDvGdvGlHCnixh\nYCSlHu7jnqWS+kf+6TGD2yBMNwhLDA2Y3YN8yaXL5ohK2BEd14fG1Wu5wg0gVP8x\nO+VLhmUo+p9qOE+mSxGj6P0cqv3FhHMXqtZ0PGZCLFWtgqfGaZiYAtok+eb/wAdv\naVPlB+11nQKBgQC/VnIN0g7/gfAYmyrRrERwmFMeXN7YsSczfqvBXMMMIisnr34p\ncTujpYcsjDawjRbdcXta1mdSmsGjDV9XkIXeR6PAHtQUiUS6AlJrpLVNV1KAgaKe\naI5MBkFmNso3rz8Zbl4+5Fn7n205nK25Trzrqj+qXQ+zgcpm7Ag9NDVw9wKBgHMK\nBp+wt4DWv0E/aGwwsM3hq3ebqDOV/ETG8uxawSfMaIuNE40myS29Lnfqi2LbZhjC\nWSVAIOJJfLlJrhetaSgjRU9WHID4EGAlU6RgWOBWgzdRTeNhvOIgvI+/vgPqhRnH\n0n5iJLuxC8jgqgo/OcntdN01P3orrTXrIFN3RaRNAoGBAIBhseLNAbMVXLO08lyM\nSnzPAdcOnl4FvdSK6+187guIs6MR9yzGsKKAPKx7AGP3pOo0j3MiN7CjPEFpukWF\nEJb8x6cQYA0GPG4nA/DYfDp+uTVqtmUGDqoLhH94xzBXgOE88vS+kUMlh5/aFM1S\nJtD2H1eB8JSJrlZvHeonj/Jx\n-----END PRIVATE KEY-----\n",
+  "client_email": "icodeidea@telzho-430515.iam.gserviceaccount.com",
+  "client_id": "105989590052836790986",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/icodeidea%40telzho-430515.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}

src/data_sources/coin_gecko.py

@@ -0,0 +1,266 @@
+import os
+from dotenv import load_dotenv
+from src.databases.redis import REDIS_CACHED
+from src.libs.constants import ONE_HOUR_IN_SECONDS, ONE_MINUTE_IN_SECONDS, ONE_MONTH_IN_SECONDS
+from src.libs.logger import logger
+from coingecko import CoinGeckoProClient, CoinGeckoDemoClient
+
+load_dotenv()
+
+redis_cache = REDIS_CACHED
+
+class CoinGecko:
+    def __init__(self, pro_api: bool = False) -> None:
+        """
+        Initializes the CoinGecko object.
+
+        Args:
+            pro_api (bool, optional): A boolean value indicating whether to use the Pro API. Defaults to False.
+
+        Returns:
+            None
+
+        Raises:
+            ValueError: If the `pro_api` argument is not a boolean value.
+
+        Note:
+            This method initializes the CoinGecko object by creating an instance of either CoinGeckoProClient or CoinGeckoDemoClient based on the `pro_api` argument. If `pro_api` is `True`, it uses the Pro API with the provided API key; otherwise, it uses the Demo API.
+        """
+        if not isinstance(pro_api, bool):
+            raise ValueError("The `pro_api` argument must be a boolean value.")
+
+        if pro_api:
+            self.cgc = CoinGeckoProClient(api_key=os.getenv('COINGECKO_PRO_API_KEY'))
+        else:
+            self.cgc = CoinGeckoDemoClient(api_key=os.getenv('COINGECKO_DEMO_API_KEY'))
+    
+    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
+    @logger.instrument()
+    def get_coin_price(self, ids: list[str], vs_currencies: list[str], cache_ttl: int = None) -> dict:
+        """
+        Retrieves the current price of a list of coins in a specified currency.
+
+        Args:
+            ids (list[str]): A list of unique identifiers for the coins to retrieve prices for.
+            vs_currencies (list[str]): A list of currency codes to convert the coin prices to.
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing the current prices of the specified coins in the specified currencies.
+
+        Example:
+            To get the current prices of a list of coins in USD, you can call:
+
+            ```python
+            coin_prices = coin_gecko.get_coin_price(ids=['bitcoin', 'ethereum'], vs_currencies=['usd'])
+            ```
+
+        Note:
+            The `get_coin_price` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MINUTE_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.simple.get_price(ids=ids, vs_currencies=vs_currencies)
+
+        return result
+    
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @logger.instrument()
+    def get_coin_data(self, id: str, cache_ttl: int = None) -> dict:
+        """
+        Retrieves detailed information about a specific coin.
+
+        Args:
+            id (str): The unique identifier of the coin.
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing the detailed information about the specified coin.
+
+        Example:
+            To get data for the coin with id 'bitcoin', you can call:
+
+            ```python
+            coin_data = coin_gecko.get_coin_data(id='bitcoin')
+            ```
+
+        Note:
+            The `get_coin_data` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.coins.get_id(id=id, localization=False, market_data=False, tickers=False, sparkline=False)
+        logger.debug(f"Result: {result}")
+
+        return result
+    
+    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
+    @logger.instrument()
+    def get_coin_category_data(self, params: dict = None, cache_ttl: int = None) -> dict:
+        """
+        Retrieves data for all available coin categories.
+
+        Args:
+            params (dict, optional): A dictionary containing optional parameters for the request. Defaults to None.
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing the data for all available coin categories.
+
+        Example:
+            To get data for all available coin categories, you can call:
+
+            ```python
+            category_data = coin_gecko.get_coin_category_data()
+            ```
+
+        Note:
+            The `get_coin_category_data` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MONTH_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        params = params or {
+            "order": "market_cap_desc",
+        }
+
+        result = self.cgc.categories.get_data(endpoint="coins/categories", params=params)
+        logger.debug(f"Result: {result}")
+
+        return { "categories": result}
+    
+    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
+    @logger.instrument()
+    def get_exchanges_list(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves a list of all available exchanges.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of all available exchanges.
+
+        Example:
+            To get a list of all available exchanges, you can call:
+
+            ```python
+            exchange_list = coin_gecko.get_exchanges_list()
+            ```
+
+        Note:
+            The `get_exchanges_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MONTH_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.exchanges.get_list()
+        logger.debug(f"Result: {result}")
+
+        return { 'exchanges': result }
+    
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @logger.instrument()
+    def get_exchange_data(self, id: str = None, cache_ttl: int = None) -> dict:
+        """
+        Retrieves detailed information about a specific exchange.
+
+        Args:
+            id (str, optional): The unique identifier of the exchange. Defaults to None.
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing the detailed information about the specified exchange.
+
+        Raises:
+            ValueError: If the `id` is not provided and `None` is passed as the argument.
+
+        Example:
+            To get data for the exchange with id 'binance', you can call:
+
+            ```python
+            exchange_data = coin_gecko.get_exchange_data(id='binance')
+            ```
+
+        Note:
+            The `get_exchange_data` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        if id is None:
+            raise ValueError("Exchange ID must be provided.")
+
+        result = self.cgc.exchanges.get_id(id=id)
+        logger.debug(f"Result: {result}")
+
+        return result
+    
+    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
+    @logger.instrument()
+    def get_coin_categories_list(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves a list of all available coin categories.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of all available coin categories.
+
+        Example:
+            To get a list of all available coin categories, you can call:
+
+            ```python
+            category_list = coin_gecko.get_coin_categories_list()
+            ```
+
+        Note:
+            The `get_coin_categories_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_MONTH_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.categories.get()
+        logger.debug(f"Result: {result}")
+
+        return { 'exchanges': result }
+    
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @logger.instrument()
+    def get_asset_platforms_list(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves a list of all available asset platforms.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of all available asset platforms.
+
+        Example:
+            To get a list of all available asset platforms, you can call:
+
+            ```python
+            asset_platforms_list = coin_gecko.get_asset_platforms_list()
+            ```
+
+        Note:
+            The `get_asset_platforms_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.asset_platform.get()
+        logger.debug(f"Result: {result}")
+
+        return { 'exchanges': result }
+    
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @logger.instrument()
+    def get_trending_coin_list(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves a list of query trending 15 coins, 7 nfts and 5 categories on CoinGecko in the last 24 hours.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (in seconds) for the cached data. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of trending coins.
+
+        Example:
+            To get a list of trending coins, you can call:
+
+            ```python
+            trending_coins = coin_gecko.get_trending_coin_list()
+            ```
+
+        Note:
+            The `get_trending_coin_list` method uses the `redis_cache` decorator to cache the result for a specified time (`ONE_HOUR_IN_SECONDS`). This means that if the same request is made within the cache time, the cached result will be returned instead of making a new API call.
+        """
+        result = self.cgc.search.get_trending()
+        logger.debug(f"Result: {result}")
+
+        return { 'exchanges': result }
+    

src/data_sources/cryptocompare.py

@@ -0,0 +1,325 @@
+import os
+from dotenv import load_dotenv
+load_dotenv()
+
+import httpx
+from enum import Enum
+from src.libs.logger import logger
+from src.databases.redis import REDIS_CACHED
+from src.libs.helper_functions import convert_to_snakecase
+from src.libs.constants import (
+    ONE_HOUR_IN_SECONDS,
+    ONE_MONTH_IN_SECONDS,
+    ONE_MINUTE_IN_SECONDS,
+    ONE_QUARTER_IN_SECONDS,
+    CRYPTO_COMPARE_API_BASE_URL,
+    CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL,
+)
+
+redis_cache = REDIS_CACHED
+
+class CryptoCompareNewsSortOrder(Enum):
+    LATEST = "latest"
+    POPULAR = "popular"
+
+    def __str__(self):
+        return self.value
+    
+class CryptoCompareLanguages(Enum):
+    English = "EN"
+    French = "FR"
+    Espanol = "ES"
+    Turkish = "TR"
+    Portuguese = "PT"
+
+    def __str__(self):
+        return self.value
+
+class CryptoCompare:
+    def __init__(self, base_url: str = None, asset_data_base_url: str = None) -> None:
+        self.CRYPTO_COMPARE_API_BASE_URL = base_url or CRYPTO_COMPARE_API_BASE_URL
+        self.CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL = asset_data_base_url or CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL
+
+    # @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
+    @logger.instrument()
+    def get_all_coins(self) -> dict:
+        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/all/coinlist"
+        logger.debug(url)
+        params = {"api_key": os.getenv("CRYPTOCOMPARE_API_KEY")}
+        logger.debug(params)
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        logger.debug(headers)
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return convert_to_snakecase(response.json())
+        except httpx.HTTPError as e:
+            print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
+    @logger.instrument()
+    def get_coin_price(self, ids: list[str], vs_currencies: list[str], cache_ttl: int = None) -> dict:
+        
+        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/pricemulti"
+        # logger.debug(url)
+
+        params = {
+            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
+            "fsyms": ",".join(ids),
+            "tsyms": ",".join(vs_currencies)
+        }
+        # logger.debug(f"{params}")
+
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        # logger.debug(f"{headers}")
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return response.json()
+        except httpx.HTTPError as e:
+            # logger.debug(f"An error occurred while making the request: {e}")
+            print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_QUARTER_IN_SECONDS)
+    @logger.instrument()
+    def get_overall_coin_data(self, symbol: str, cache_ttl: int = None) -> dict:
+        
+        url = f"{self.CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL}data/by/symbol"
+        logger.debug(f"Query URL: {url}")
+
+        params = {
+            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
+            "asset_symbol": symbol,
+        }
+        logger.debug(f"Query params: {params}")
+
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        logger.debug(f"Query headers {headers}")
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return response.json()
+        except httpx.HTTPError as e:
+            logger.debug(f"An error occurred while making the request: {e}")
+            # print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_QUARTER_IN_SECONDS)
+    @logger.instrument()
+    def get_news_categories(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves the news categories from the CryptoCompare API.
+       
+        Args:
+            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing the news categories from the CryptoCompare API.
+
+        Raises:
+            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.
+        """
+        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/news/categories"
+        logger.debug(f"Query URL: {url}")
+
+        params = {
+            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
+        }
+        logger.debug(f"Query params: {params}")
+
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        logger.debug(f"Query headers {headers}")
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return response.json()
+        except httpx.HTTPError as e:
+            logger.debug(f"An error occurred while making the request: {e}")
+            # print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @logger.instrument()
+    def get_latest_news_articles(
+        self,
+        language: CryptoCompareLanguages = CryptoCompareLanguages.English,
+        sort_order: CryptoCompareNewsSortOrder = CryptoCompareNewsSortOrder.LATEST,
+        cache_ttl: int = None
+    ) -> dict:
+        """
+        Retrieves the latest news articles from the CryptoCompare API.
+
+        Args:
+            language (CryptoCompareLanguages, optional): The language of the news articles. Defaults to English.
+            sort_order (CryptoCompareNewsSortOrder, optional): The sort order of the news articles. Defaults to 'latest'.
+            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing the latest news articles from the CryptoCompare API.
+
+        Raises:
+            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.
+
+        Example:
+            # Get the latest news articles in English, sorted by popularity
+            latest_news = crypto_compare_instance.get_latest_news_articles(language=CryptoCompareLanguages.English, sort_order=CryptoCompareNewsSortOrder.POPULAR)
+        """
+        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/v2/news/"
+        logger.debug(f"Query URL: {url}")
+
+        params = {
+            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
+            "lang": language,
+            "sortOrder": sort_order,
+        }
+        logger.debug(f"Query params: {params}")
+
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        logger.debug(f"Query headers {headers}")
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return response.json()
+        except httpx.HTTPError as e:
+            logger.debug(f"An error occurred while making the request: {e}")
+            # print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
+    @logger.instrument()
+    def get_all_exchanges_general_info(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves general information about all supported exchanges from the CryptoCompare API.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing general information about all supported exchanges from the CryptoCompare API.
+
+        Raises:
+            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.
+
+        Example:
+            # Get general information about all supported exchanges
+            all_exchanges_info = crypto_compare_instance.get_all_exchanges_general_info()
+        """
+        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/exchanges/general"
+        logger.debug(f"Query URL: {url}")
+
+        params = {
+            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
+        }
+        logger.debug(f"Query params: {params}")
+
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        logger.debug(f"Query headers {headers}")
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return response.json()
+        except httpx.HTTPError as e:
+            logger.debug(f"An error occurred while making the request: {e}")
+            # print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
+    @logger.instrument()
+    def get_all_wallets_general_info(self, cache_ttl: int = None) -> dict:
+        """
+        Retrieves general information about all supported wallets from the CryptoCompare API.
+
+        Args:
+            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing general information about all supported wallets from the CryptoCompare API.
+
+        Raises:
+            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.
+
+        Example:
+            # Get general information about all supported wallets
+            all_wallets_info = crypto_compare_instance.get_all_wallets_general_info()
+        """
+        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/wallets/general"
+        logger.debug(f"Query URL: {url}")
+
+        params = {
+            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
+        }
+        logger.debug(f"Query params: {params}")
+
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        logger.debug(f"Query headers {headers}")
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return response.json()
+        except httpx.HTTPError as e:
+            logger.debug(f"An error occurred while making the request: {e}")
+            # print(f"An error occurred while making the request: {e}")
+            return None
+        
+    # @redis_cache(ttl=ONE_MONTH_IN_SECONDS)
+    @logger.instrument()
+    def get_top_tier_exchanges_list(
+        self,
+        top_tier: bool = False,
+        cache_ttl: int = None
+    ) -> dict:
+        """
+        Retrieves a list of top-tier exchanges from the CryptoCompare API.
+
+        Args:
+            top_tier (bool, optional): A boolean flag indicating whether to retrieve top-tier exchanges. Defaults to False.
+            cache_ttl (int, optional): The time-to-live (TTL) for the cached response. Defaults to None.
+
+        Returns:
+            dict: A dictionary containing a list of top-tier exchanges from the CryptoCompare API.
+
+        Raises:
+            httpx.HTTPError: If an error occurs while making the request to the CryptoCompare API.
+
+        Example:
+            # Get a list of top-tier exchanges
+            top_tier_exchanges = crypto_compare_instance.get_top_tier_exchanges_list(top_tier=True)
+        """
+        url = f"{self.CRYPTO_COMPARE_API_BASE_URL}data/v4/all/exchanges"
+        logger.debug(f"Query URL: {url}")
+
+        params = {
+            "api_key": os.getenv("CRYPTOCOMPARE_API_KEY"),
+            "topTier": 'true' if top_tier else 'false',
+        }
+        logger.debug(f"Query params: {params}")
+
+        headers = {"Content-type":"application/json; charset=UTF-8"}
+        logger.debug(f"Query headers {headers}")
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(url, params=params, headers=headers)
+                response.raise_for_status()  # Raise an exception if the request was unsuccessful
+                return response.json()
+        except httpx.HTTPError as e:
+            logger.debug(f"An error occurred while making the request: {e}")
+            # print(f"An error occurred while making the request: {e}")
+            return None
+        

src/data_sources/dexscreener.py

@@ -0,0 +1,106 @@
+from dotenv import load_dotenv
+
+import httpx
+from typing import List, Union
+from src.databases.redis import REDIS_CACHED
+from src.libs.constants import ONE_HOUR_IN_SECONDS, ONE_MINUTE_IN_SECONDS
+from src.libs.constants import DEX_SCREENER_BASE_URL
+
+load_dotenv()
+
+redis_cache = REDIS_CACHED
+
+class DexScreener:
+    """
+    A class for interacting with the Dex Screener API.
+
+    Attributes:
+        DEX_SCREENER_BASE_URL (str): The base URL for the Dex Screener API.
+
+    Methods:
+        __init__(self, base_url: str) -> None:
+            Initialize the DexScreener class.
+
+        get_pairs(self, chain_id: str, pair_addresses: str) -> dict:
+
+        get_tokens(self, token_addresses: Union[str, List[str]]) -> dict:
+
+        search_pairs(self, query: str) -> dict:
+    """
+    def __init__(self, base_url: str = None) -> None:
+        self.DEX_SCREENER_BASE_URL = base_url or DEX_SCREENER_BASE_URL
+
+    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
+    def get_pairs(self, chain_id: str, pair_addresses: str) -> dict:
+        """
+        This method is used to fetch pair data from Dex Screener API based on the provided chain_id and pair_addresses.
+
+        Parameters:
+        chain_id (str): The ID of the blockchain network.
+        pair_addresses (str): The address(es) of the pair(s) on the blockchain.
+
+        Returns:
+        dict: A dictionary containing the JSON response from the API. Returns None if an error occurs.
+
+        Raises:
+        httpx.HTTPError: If an error occurs while making the request to the API.
+        """
+        url = f"{self.DEX_SCREENER_BASE_URL}pairs/{chain_id}/{pair_addresses}"
+        try:
+            response = httpx.get(url)
+            response.raise_for_status()  # Raise an exception if the request was unsuccessful
+            return response.json()
+        except httpx.HTTPError as e:
+            print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
+    def get_tokens(self, token_addresses: Union[str, List[str]]) -> dict:
+        """
+        This method is used to fetch token data from Dex Screener API based on the provided token_addresses.
+
+        Parameters:
+        token_addresses (Union[str, List[str]]): The address(es) of the token(s) on the blockchain.
+            This parameter can be a single address (str) or a list of addresses (List[str]).
+
+        Returns:
+        dict: A dictionary containing the JSON response from the API. Returns None if an error occurs.
+
+        Raises:
+        httpx.HTTPError: If an error occurs while making the request to the API.
+        """
+        if isinstance(token_addresses, list):
+            token_addresses = ','.join(token_addresses)
+
+        url = f"{self.DEX_SCREENER_BASE_URL}tokens/{token_addresses}"
+        try:
+            response = httpx.get(url)
+            response.raise_for_status()  # Raise an exception if the request was unsuccessful
+            return response.json()
+        except httpx.HTTPError as e:
+            print(f"An error occurred while making the request: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    def search(self, query: str) -> dict:
+        """
+        This method is used to search for pairs matching the provided query from Dex Screener API.
+        The query may include pair address, token address, token name, or token symbol.
+
+        Parameters:
+        query (str): The search query.
+
+        Returns:
+        dict: A dictionary containing the JSON response from the API. Returns None if an error occurs.
+
+        Raises:
+        httpx.HTTPError: If an error occurs while making the request to the API.
+        """
+        url = f"{self.DEX_SCREENER_BASE_URL}search/?q={query}"
+        try:
+            response = httpx.get(url)
+            response.raise_for_status()  # Raise an exception if the request was unsuccessful
+            return response.json()
+        except httpx.HTTPError as e:
+            print(f"An error occurred while making the request: {e}")
+            return None

src/databases/postgres.py

@@ -0,0 +1,4 @@
+import os
+from sqlalchemy import create_engine
+
+sqlalchemy_engine = create_engine(url=os.getenv('POSTGRES_CONNECTION_STRING'))

src/databases/redis.py

@@ -0,0 +1,132 @@
+import os
+import json
+from dotenv import load_dotenv
+from upstash_redis import Redis
+from src.libs.logger import logger
+from src.libs.helper_functions import chunk_data, create_uuid_from_string
+load_dotenv()
+
+url = os.getenv("UPSTASH_REDIS_REST_URL")
+token = os.getenv("UPSTASH_REDIS_REST_TOKEN")
+
+REDIS = Redis(url=url, token=token)
+REDIS_PIPELINE = REDIS.pipeline()
+
+
+@logger.instrument()
+def args_to_key(*args, **kwargs):
+    """
+    This function generates a unique key based on the provided arguments and keyword arguments.
+
+    Args:
+        *args: A variable number of positional arguments.
+        **kwargs: A variable number of keyword arguments.
+
+    Returns:
+        str: A unique key string generated from the provided arguments and keyword arguments.
+
+    Example:
+        >>> args_to_key("hello", "world", a=1, b=2)
+        'hello_world_a_1_b_2'
+    """
+    params = []
+
+    # TODO: turn these into debugging logs
+    logger.debug(f"args: {args}")
+    logger.debug(f"kwargs: {kwargs}")
+
+    # Append the names of callable arguments to the params list
+    for arg in args:
+        if callable(arg):
+            params.append(arg.__name__)
+
+    # Append the string representations of non-callable arguments and keyword arguments to the params list
+    for arg in args:
+        if not callable(arg):
+            params.append(cast_args_to_string_and_return_first_index(arg))
+
+    for kwarg in kwargs.values():
+        params.append(str(kwarg))
+
+    # Join the elements in the params list using the '_' character as a separator
+    return "_".join(params)
+
+
+@logger.instrument()
+def REDIS_CACHED(ttl: int = 3600, chunk: bool = False):
+    """
+    This decorator caches the result of a function call in Redis.
+
+    Args:
+        ttl (int): The time-to-live (in seconds) for the cached result. Defaults to 3600 seconds (1 hour).
+        chunk (bool): Whether to chunk the result of the original function call. Defaults to False.
+
+    Returns:
+        A wrapper function that caches the result of the original function call.
+
+    Example:
+        >>> @REDIS_CACHED(ttl=60,  chunk=True)
+        ... def example_function(arg1, arg2):
+        ...     return arg1 + arg2
+        ...
+        ... cached_result = example_function(3, 4)
+        ... print(cached_result)  # Output: 7
+        ... print(cached_result)  # Output: 7 (from cache)
+    """
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            r = REDIS
+
+            cache_key = args_to_key(func, *args, **kwargs)
+            logger.debug(f"Cache key: {cache_key}") # TODO: turn these into debugging logs
+            cache_key = str(create_uuid_from_string(cache_key))
+            logger.debug(f"Cache key: {cache_key}") # TODO: turn these into debugging logs
+
+            # Test if a matching cache key exists
+            cached = r.get(cache_key)
+            if cached:
+                # Found in cache, return it
+                return json.loads(cached)
+
+            # Otherwise, pass everything to the downstream function
+            result = func(*args, **kwargs)
+
+            # Set cache time-to-live duration
+            # Use the default TTL if not provided as an argument
+            cache_ttl = kwargs.get('cache_ttl')
+            ttl_seconds = cache_ttl or kwargs.get('ttl', ttl)
+
+            if chunk:
+                chunked_result = chunk_data(result.data, 100)
+                for i in range(len(chunked_result)):
+                    r.rpush(cache_key, json.dumps(chunked_result[i]))
+                r.pexpire(cache_key, ttl_seconds)
+            else:
+                # Put the result from downstream function into cache, with a TTL
+                # So next call with the same parameters will be handled by the cache
+                r.setex(cache_key, ttl_seconds, result)
+
+            # Return the result transparently
+            return result
+        return wrapper
+    return decorator
+
+
+@logger.instrument()
+def cast_args_to_string_and_return_first_index(args):
+    """
+    This function takes an argument, casts it to a string, removes '<' and '>', splits it by spaces, and returns the first index.
+
+    Args:
+        args (any): The argument to be processed.
+
+    Returns:
+        str or None: The first index of the argument after processing, or None if the argument is empty.
+
+    Example:
+        >>> cast_args_to_string_and_return_first_index("hello <world>")
+        'hello'
+    """
+    args_str = str(args).strip('<>').split(' ')
+    return args_str[0] if args_str else None
+

src/knowledge_bases/combined.py

@@ -0,0 +1,60 @@
+from phi.knowledge.combined import CombinedKnowledgeBase
+from phi.vectordb.pgvector import PgVector2
+from src.knowledge_bases.json import json_knowledge_base
+from src.knowledge_bases.pdf import pdf_knowledge_base
+from phi.knowledge.pdf import PDFUrlKnowledgeBase
+from phi.knowledge.website import WebsiteKnowledgeBase
+from src.databases.postgres import sqlalchemy_engine
+from src.libs.helper_functions import get_pdf_urls, get_pdf_urls_with_root_url
+from src.libs.google_drive import get_drive_file_urls, get_pdf_urls_with_root_url
+from phi.embedder.ollama import OllamaEmbedder
+
+# Configuration for Google Drive API
+# service_account_file = 'src/config/telzho-430515-c2622883c30e.json'
+# folder_id = '1veFNWFu68qO9dDPxouLSdB2pHGunosar'
+
+# Get file URLs from Google Drive folder
+# pdf_urls = get_drive_file_urls(service_account_file, folder_id)
+
+# pdf_urls = get_pdf_urls_with_root_url(
+#     "https://swiftlynxtechnologies.com/Rhapsody/",
+#     "https://swiftlynxtechnologies.com/Rhapsody"
+# )
+# print(pdf_urls)
+
+pdf_urls = get_pdf_urls_with_root_url("https://swiftlynxtechnologies.com/Rhapsody/")
+print(pdf_urls)
+
+knowledge_base = CombinedKnowledgeBase(
+    sources=[
+        json_knowledge_base,
+        # pdf_knowledge_base
+        # PDFUrlKnowledgeBase(
+        #     urls=pdf_urls,
+        #     # Table name: pastor_chris_library
+        #     vector_db=PgVector2(
+        #         collection="pastor_chris_library",
+        #         db_engine=sqlalchemy_engine,
+        #         embedder=OllamaEmbedder()
+        #     ),
+        # ),
+        # WebsiteKnowledgeBase(
+        #     urls=[''],
+        #     # Number of links to follow from the seed URLs
+        #     max_links=10,
+        #     # Table name: pastor_chris_library
+        #     vector_db=PgVector2(
+        #         collection="pastor_chris_library",
+        #         db_engine=sqlalchemy_engine,
+        #     ),
+        # ),
+        # website_knowledge_base,
+        # local_pdf_knowledge_base,
+    ],
+    vector_db=PgVector2(
+        # Table name: ai.combined_documents
+        collection="combined_documents",
+        db_engine=sqlalchemy_engine,
+        embedder=OllamaEmbedder()
+    ),
+)

src/knowledge_bases/json.py

@@ -0,0 +1,63 @@
+import json
+from fs_s3fs import S3FS
+from src.libs.logger import logger
+from src.libs.s3fs import get_s3_credentials
+from phi.vectordb.pgvector import PgVector2
+from phi.knowledge.json import JSONKnowledgeBase
+from src.databases.postgres import sqlalchemy_engine
+from phi.embedder.ollama import OllamaEmbedder
+
+class JSONKnowledgeBaseExtended(JSONKnowledgeBase):
+    s3fs: S3FS = None  # Explicitly declare the s3fs attribute
+
+    def __init__(
+        self, 
+        s3_bucket_name, 
+        vector_db, 
+        s3_access_key_id,
+        s3_secret_access_key,
+        s3_endpoint_url,
+        s3_region,
+    ):
+        super().__init__(path=s3_bucket_name, vector_db=vector_db, bucket_name = s3_bucket_name)
+
+        # Initialize the S3 filesystem
+        self.s3fs = S3FS(
+            bucket_name=s3_bucket_name,
+            aws_access_key_id=s3_access_key_id,
+            aws_secret_access_key=s3_secret_access_key,
+            endpoint_url = s3_endpoint_url,
+            region = s3_region,
+        )
+
+    def load_knowledge_base(self, recreate: bool = False):
+        json_knowledge_base.load(recreate=recreate)
+
+    def store_json_data_in_s3(self, json_data, file_path):
+        if file_path[0] == '/':
+            file_path = f"/json-data/{file_path[1:]}"
+        else:
+            file_path = f"/json-data/{file_path}"
+
+        logger.info(f"Storing JSON data in S3 bucket: {self.s3fs._bucket_name} at path: {file_path}")
+        
+        # Open the file in write mode and write the JSON data
+        self.s3fs.open(path = f"/{file_path}", mode = 'w').write(json.dumps(json_data, indent=2))
+        return True
+
+
+# S3 credentials
+_s3_credendtials = get_s3_credentials()
+_json_knowledge_base_arguments = {
+    "vector_db": PgVector2(
+        collection="json_documents",
+        db_engine=sqlalchemy_engine,
+        embedder=OllamaEmbedder()
+    ),
+    **_s3_credendtials
+}
+
+# Initialize the extended JSONKnowledgeBase with the S3 bucket name and S3 credentials
+json_knowledge_base = JSONKnowledgeBaseExtended(
+    **_json_knowledge_base_arguments
+)

src/knowledge_bases/pdf.py

@@ -0,0 +1,57 @@
+import json
+from fs_s3fs import S3FS
+from src.libs.logger import logger
+from src.libs.s3fs import get_s3_credentials
+from phi.vectordb.pgvector import PgVector2
+from phi.knowledge.json import JSONKnowledgeBase
+from phi.knowledge.pdf import PDFUrlKnowledgeBase
+from src.databases.postgres import sqlalchemy_engine
+from phi.embedder.ollama import OllamaEmbedder
+
+
+class PDFUrlKnowledgeBaseExtended(PDFUrlKnowledgeBase):
+    s3fs: S3FS = None  # Explicitly declare the s3fs attribute
+
+    def __init__(
+            self,
+            s3_bucket_name,
+            vector_db,
+            s3_access_key_id,
+            s3_secret_access_key,
+            s3_endpoint_url,
+            s3_region,
+    ):
+        super().__init__(path=s3_bucket_name, vector_db=vector_db, bucket_name=s3_bucket_name)
+
+        # Initialize the S3 filesystem
+        self.s3fs = S3FS(
+            bucket_name=s3_bucket_name,
+            aws_access_key_id=s3_access_key_id,
+            aws_secret_access_key=s3_secret_access_key,
+            endpoint_url=s3_endpoint_url,
+            region=s3_region,
+        )
+
+    def load_knowledge_base(self, recreate: bool = False):
+        pdf_knowledge_base.load(recreate=recreate)
+
+    def chunk_and_store_to_vector_db(self, urls):
+        pdf_knowledge_base.urls = urls
+
+
+# S3 credentials
+_s3_credendtials = get_s3_credentials()
+_pdf_knowledge_base_arguments = {
+    "vector_db": PgVector2(
+        collection="pdf_documents",
+        db_engine=sqlalchemy_engine,
+        embedder=OllamaEmbedder()
+
+    ),
+    **_s3_credendtials
+}
+
+# Initialize the extended PDFKnowledgeBase with the S3 bucket name and S3 credentials
+pdf_knowledge_base = PDFUrlKnowledgeBaseExtended(
+    **_pdf_knowledge_base_arguments
+)

src/knowledge_bases/pipeline.py

@@ -0,0 +1,48 @@
+from chainlit import make_async
+from src.data_sources.cryptocompare import CryptoCompare
+from src.data_sources.coin_gecko import CoinGecko
+from src.knowledge_bases.json import json_knowledge_base
+
+coin_gecko = CoinGecko()
+crypto_compare = CryptoCompare()
+
+store_json_data_in_s3 = make_async(json_knowledge_base.store_json_data_in_s3)
+
+async def fetch_and_load_crypto_compare_json_data():
+    get_all_coins = make_async(crypto_compare.get_all_coins)
+    get_news_categories = make_async(crypto_compare.get_news_categories)
+    get_top_tier_exchanges_list = make_async(crypto_compare.get_top_tier_exchanges_list)
+    get_all_wallets_general_info = make_async(crypto_compare.get_all_wallets_general_info)
+    get_all_exchanges_general_info = make_async(crypto_compare.get_all_exchanges_general_info)
+
+    all_coins = await get_all_coins()
+    news_categories = await get_news_categories()
+    top_tier_exchanges_list = await get_top_tier_exchanges_list()
+    all_wallets_general_info = await get_all_wallets_general_info()
+    all_exchanges_general_info = await get_all_exchanges_general_info()
+
+    await store_json_data_in_s3(all_coins, "crypto_compare_all_coins.json")
+    await store_json_data_in_s3(news_categories, "crypto_compare_news_categories.json")
+    await store_json_data_in_s3(top_tier_exchanges_list, "crypto_compare_top_tier_exchanges_list.json")
+    await store_json_data_in_s3(all_wallets_general_info, "crypto_compare_all_wallets_general_info.json")
+    await store_json_data_in_s3(all_exchanges_general_info, "crypto_compare_all_exchanges_general_info.json")
+
+async def fetch_and_load_coin_gecko_json_data():
+    get_all_coins = make_async(coin_gecko.get_exchanges_list)
+    get_asset_platforms_list = make_async(coin_gecko.get_asset_platforms_list)
+
+    all_exchange_data = await get_all_coins()
+    asset_platforms_list = await get_asset_platforms_list()
+
+    await store_json_data_in_s3(all_exchange_data, "coin_gecko_all_exchange_data")
+    await store_json_data_in_s3(asset_platforms_list, "coin_gecko_asset_platforms_list")
+
+async def load_json_data_into_knowledge_base():
+    load_knowledge_base = make_async(json_knowledge_base.load_knowledge_base)
+
+    await load_knowledge_base(recreate=True)
+
+async def main():
+    await fetch_and_load_coin_gecko_json_data()
+    await fetch_and_load_crypto_compare_json_data()
+    await load_json_data_into_knowledge_base()

src/knowledge_bases/store_data.py

@@ -0,0 +1,10 @@
+from src.data_sources.cryptocompare import CryptoCompare
+from src.libs.helper_functions import store_json_data
+from src.libs.logger import logger
+
+ccp = CryptoCompare()
+
+def store_coin_data(id: str):
+    result = ccp.get_overall_coin_data(symbol=id)
+    store_json_data(result, f"data/coin_data/{id}.json")
+    logger.debug(f"Data stored for {id}")

src/libs/constants.py

@@ -0,0 +1,70 @@
+"""
+Constants representing the number of seconds in a quarter.
+"""
+ONE_QUARTER_IN_SECONDS : int = (86400 * 30) * 3
+
+"""
+Constants representing the number of seconds in a month.
+"""
+ONE_MONTH_IN_SECONDS : int = 86400 * 30
+
+"""
+Constants representing the number of seconds in a week.
+"""
+ONE_WEEK_IN_SECONDS : int = 604800
+
+"""
+Constants representing the number of seconds in a day.
+"""
+ONE_DAY_IN_SECONDS : int = 86400
+
+"""
+Constant representing the number of seconds in one hour.
+"""
+ONE_HOUR_IN_SECONDS : int = 3600
+
+"""
+Constant representing the number of seconds in one minute.
+"""
+ONE_MINUTE_IN_SECONDS : int = 60
+
+"""
+This constant is used to represent the base URL for DexScreener API.
+
+DEX_SCREENER_BASE_URL : str
+"""
+DEX_SCREENER_BASE_URL : str = "https://api.dexscreener.com/latest/dex/"
+
+"""
+This constant is used to represent the base endpoint for Jina Search API.
+
+JINA_SEARCH_BASE_ENDPOINT : str
+"""
+JINA_SEARCH_BASE_ENDPOINT : str = "https://s.jina.ai/"
+
+"""
+This constant is used to represent the base endpoint for Jina Reader API.
+
+JINA_READER_BASE_ENDPOINT : str
+"""
+JINA_READER_BASE_ENDPOINT : str = "https://r.jina.ai/"
+
+"""
+This constant is used to represent the base URL for CryptoCompare API.
+
+CRYPTO_COMPARE_BASE_URL: str
+"""
+CRYPTO_COMPARE_API_BASE_URL: str = "https://min-api.cryptocompare.com/"
+
+
+"""
+This constant is used to represent the base URL for CryptoCompare  asset data API.
+
+CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL: str
+"""
+CRYPTO_COMPARE_ASSET_DATA_API_BASE_URL : str = "https://data-api.cryptocompare.com/asset/v1/"
+
+
+# TTLs
+
+SEARCH_DATA_TTL : int = ONE_WEEK_IN_SECONDS

src/libs/google_drive.py

@@ -0,0 +1,54 @@
+import json
+from google.oauth2 import service_account
+from googleapiclient.discovery import build
+import requests
+from bs4 import BeautifulSoup
+
+
+def get_pdf_urls_with_root_url(root_url):
+    """
+    Returns an array of URLs of PDF files in the specified remote directory.
+
+    Parameters:
+    root_url (str): The root URL to the directory listing.
+
+    Returns:
+    list: A list of URLs to the PDF files in the directory.
+    """
+    pdf_urls = []
+
+    # Make a request to get the content of the directory
+    response = requests.get(root_url)
+
+    if response.status_code == 200:
+        # Parse the HTML content
+        soup = BeautifulSoup(response.content, 'html.parser')
+
+        # Iterate over all links in the directory listing
+        for link in soup.find_all('a'):
+            href = link.get('href')
+            if href and href.lower().endswith('.pdf'):
+                # Construct the full URL for the PDF file
+                file_url = root_url.rstrip('/') + '/' + href.lstrip('/')
+                pdf_urls.append(file_url)
+    else:
+        print(f"Failed to access {root_url}, status code: {response.status_code}")
+
+    return pdf_urls
+
+
+def get_drive_file_urls(service_account_file, folder_id):
+    credentials = service_account.Credentials.from_service_account_file(
+        service_account_file,
+        scopes=['https://www.googleapis.com/auth/drive.readonly'],
+    )
+
+    service = build('drive', 'v3', credentials=credentials)
+
+    query = f"'{folder_id}' in parents"
+    results = service.files().list(q=query, fields="files(id, webViewLink)").execute()
+    items = results.get('files', [])
+
+    file_urls = [item['webViewLink'] for item in items]
+
+    return file_urls

src/libs/helper_functions.py

@@ -0,0 +1,215 @@
+from ulid import ULID
+from typing import Dict
+import os, re, uuid, hashlib
+from dotenv import load_dotenv
+
+load_dotenv()
+
+def get_pdf_urls(folder_path):
+    """
+    Returns an array of URLs of PDF files in the specified folder.
+
+    Parameters:
+    folder_path (str): The path to the folder.
+
+    Returns:
+    list: A list of URLs to the PDF files in the folder.
+    """
+    pdf_urls = []
+
+    # Iterate over all files in the specified folder
+    for filename in os.listdir(folder_path):
+        # Check if the file is a PDF
+        if filename.lower().endswith('.pdf'):
+            # Construct the URL for the PDF file
+            file_url = os.path.join(folder_path, filename)
+            pdf_urls.append(file_url)
+
+    return pdf_urls
+
+
+def get_pdf_urls_with_root_url(folder_path, root_url):
+    """
+    Returns an array of URLs of PDF files in the specified folder.
+
+    Parameters:
+    folder_path (str): The path to the folder.
+    root_url (str): The root URL to construct the file URLs.
+
+    Returns:
+    list: A list of URLs to the PDF files in the folder.
+    """
+    pdf_urls = []
+
+    # Iterate over all files in the specified folder
+    for filename in os.listdir(folder_path):
+        # Check if the file is a PDF
+        if filename.lower().endswith('.pdf'):
+            # Construct the URL for the PDF file using the root URL
+            file_url = os.path.join(root_url, filename)
+            pdf_urls.append(file_url)
+
+    return pdf_urls
+
+
+# Example usage:
+# folder_path = "/path/to/your/folder"
+# print(get_pdf_urls(folder_path))
+
+
+def chunk_data(data : list | dict, chunk_size: int):
+    """
+    This function takes an array and a chunk size as input, and returns a new array
+    where the original array is divided into smaller chunks of the specified size.
+
+    Parameters:
+    data     (list): The original data to be chunked.
+    chunk_size (int): The size of each chunk.
+
+    Returns:
+    list: A new array containing the chunks of the original array.
+
+    Example:
+    >>> chunk_array([1, 2, 3, 4, 5, 6], 2)
+    [[1, 2], [3, 4], [5, 6]]
+    """
+    is_object = isinstance(data, dict)
+    is_array = isinstance(data, list)
+
+    if not is_object and not is_array:
+        raise TypeError("Data must be a list or a dictionary.")
+    elif is_array:
+        return [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]
+    elif is_object:
+        items = list(data.items())
+        for i in range(0, len(items), chunk_size):
+            yield dict(items[i:i + chunk_size])
+
+
+def generate_ulid(seed: any = None) -> str:
+    """
+    This function generates a Universally Unique Lexicographically Sortable Identifier (ULID).
+    If a seed is provided, it will be used as the basis for the ULID generation.
+
+    Parameters:
+    seed (any, optional): A value to be used as the basis for the ULID generation. Defaults to None.
+
+    Returns:
+    str: A string representing the generated ULID.
+
+    Example:
+    >>> generate_ulid()
+    '00000000-0001-0100-0000-000000000001'
+    >>> generate_ulid('example_seed')
+    '00000000-0001-0100-0000-000000000002'
+    """
+    if seed is None:
+        ulid = ULID()
+    else:
+        ulid = ULID(seed)
+
+    return ulid.generate()
+
+def create_uuid_from_string(val: str) -> str:
+    """
+    This function takes a string as input and generates a UUID (Universally Unique Identifier)
+    using the input string as the basis for the hash.
+
+    Parameters:
+    val (str): The input string from which the UUID will be generated.
+
+    Returns:
+    str: A string representing the generated UUID.
+
+    Example:
+    >>> create_uuid_from_string('example_string')
+    '00000000-0001-0100-0000-000000000001'
+    """
+    hex_string = hashlib.md5(val.encode("UTF-8")).hexdigest()
+    return str(uuid.UUID(hex=hex_string))
+
+
+def to_snake_case(s):
+    """
+    This function takes a string as input and converts it to snake_case format.
+
+    Parameters:
+    s (str): The input string to be converted to snake_case.
+
+    Returns:
+    str: A string in snake_case format.
+
+    Example:
+    >>> to_snake_case('FirstName')
+    'first_name'
+    """
+    if not s:
+        return ''
+    
+    # Check if the string is in all caps
+    if s.isupper():
+        return s
+    
+    return '_'.join(
+        word.lower() for word in re.findall(r'[A-Z]{2,}(?=[A-Z][a-z]+[0-9]*|\b)|[A-Z]?[a-z]+[0-9]*|[A-Z]|[0-9]+', s)
+    )
+
+def convert_to_snakecase(original_data):
+    """
+    This function takes a dictionary or list of dictionaries as input and converts its keys to snake_case format.
+    If the input is a list of dictionaries, it will recursively convert each nested dictionary.
+
+    Parameters:
+    original_data (dict or list): The input dictionary or list of dictionaries.
+
+    Returns:
+    dict: A new dictionary with keys in snake_case format.
+
+    Example:
+    >>> convert_to_snakecase({'FirstName': 'John', 'LastName': 'Doe'})
+    {'first_name': 'John', 'last_name': 'Doe'}
+    >>> convert_to_snakecase([{'FirstName': 'Jane', 'LastName': 'Smith'}, {'FirstName': 'Bob', 'LastName': 'Johnson'}])
+    [{'first_name': 'Jane', 'last_name': 'Smith'}, {'first_name': 'Bob', 'last_name': 'Johnson'}]
+    """
+    if isinstance(original_data, dict):
+        transformed_dict = {}
+        for k, v in original_data.items():
+            new_key = to_snake_case(k)
+            if isinstance(v, (dict, list)):
+                transformed_dict[new_key] = convert_to_snakecase(v)
+            else:
+                transformed_dict[new_key] = v
+        return transformed_dict
+    elif isinstance(original_data, list):
+        return [convert_to_snakecase(item) for item in original_data]
+    else:
+        raise TypeError("Input must be a dictionary or a list of dictionaries.")
+    
+import json
+
+def store_json_data(data, output_file):
+    try:
+        # Create directory if it doesn't exist
+        os.makedirs(os.path.dirname(output_file), exist_ok=True)
+
+        # Write the JSON data to a file
+        with open(output_file, 'w') as file:
+            json.dump(data, file, indent=4)
+        
+        print(f"Data successfully stored in {output_file}")
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+def get_headers() -> Dict[str, str]:
+    """
+    Returns the headers for the requests including the API key.
+    """
+    return {
+        '0x-api-key': f"{os.getenv('OX_API_KEY')}"
+    }
+
+def get_private_key() -> str:
+    """
+    Returns a private key.
+    """
+    return os.getenv('WALLET_PRIVATE_KEY')

src/libs/logger.py

@@ -0,0 +1,17 @@
+import os
+from dotenv import load_dotenv
+
+import logfire as lf
+
+load_dotenv()
+
+lf.configure(
+    token=os.getenv('LOGFIRE_TOKEN'),
+    # pydantic_plugin=lf.PydanticPlugin(record='all'),
+    console=lf.ConsoleOptions(min_log_level= os.getenv('LOG_LEVEL'))
+)
+lf.instrument_redis()
+lf.instrument_httpx()
+lf.instrument_requests()
+
+logger = lf

src/libs/rpc_client.py

@@ -0,0 +1,52 @@
+import httpx, uuid, os
+from typing import Optional, Union
+from dotenv import load_dotenv
+
+load_dotenv()
+
+rpc_server_url = os.getenv('GIINO_RPC_SERVER_URL')
+
+async def rpc_call(
+    method_name: str,  # The name of the RPC method to be called
+    params: Optional[Union[dict, list]] = None,  # Optional parameters for the RPC method
+    url: str = rpc_server_url  # The URL of the RPC server
+) -> dict:  # Returns the JSON response from the RPC server
+    """
+    This function makes an RPC call to the specified URL with the given method name and parameters.
+
+    Args:
+        method_name (str): The name of the RPC method to be called.
+        params (Optional[Union[dict, list]], optional): Optional parameters for the RPC method. Defaults to None.
+        url (str, optional): The URL of the RPC server. Defaults to rpc_server_url.
+
+    Returns:
+        dict: The JSON response from the RPC server.
+
+    Raises:
+        httpx.RequestError: If an error occurs while making the RPC call.
+
+    """
+    headers = {
+        'Content-Type': 'application/json',
+    }
+
+    auth = httpx.BasicAuth(
+        username=os.getenv('GIINO_RPC_SERVER_BASIC_AUTH_USERNAME'),
+        password=os.getenv('GIINO_RPC_SERVER_BASIC_AUTH_PASSWORD')
+    )
+
+    payload = {
+        'method': method_name,
+        'params': params,
+        'jsonrpc': '2.0',
+        'id': str(uuid.uuid4()),
+    }
+
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, json=payload, headers=headers, auth=auth)
+            response.raise_for_status()
+            return response.json()
+    except httpx.RequestError as e:
+        print(f"Error making RPC call: {e}")
+        return None

src/libs/s3fs.py

@@ -0,0 +1,26 @@
+import os
+import s3fs
+from fs_s3fs import S3FS
+from dotenv import load_dotenv
+
+load_dotenv()
+
+exchanges_data_s3fs = S3FS(
+    bucket_name = 'exchanges-data',
+    aws_access_key_id = os.getenv('DO_SPACES_ACCESS_KEY'),
+    aws_secret_access_key = os.getenv('DO_SPACES_SECRET_KEY'),
+    endpoint_url = os.getenv('DO_SPACES_ENDPOINT'),
+    region = os.getenv('DO_SPACES_REGION'),
+)
+
+# AWS credentials
+s3_credentials = {
+    "s3_access_key_id": os.getenv("DO_SPACES_ACCESS_KEY"),
+    "s3_secret_access_key": os.getenv("DO_SPACES_SECRET_KEY"),
+    "s3_endpoint_url": os.getenv("DO_SPACES_ENDPOINT_URL"),
+    "s3_bucket_name": os.getenv("DO_SPACES_BUCKET_NAME"),
+    "s3_region": os.getenv("DO_SPACES_REGION_NAME"),
+}
+
+def get_s3_credentials():
+    return s3_credentials

src/libs/token_approval_helper.py

@@ -0,0 +1,61 @@
+import requests
+from web3 import Web3
+from typing import Dict, Any
+
+class TokenApprovalHelper:
+    def __init__(self, web3: Web3, private_key: str):
+        self.web3 = web3
+        self.private_key = private_key
+
+    def approve_token(self, erc20_contract_address: str, exchange_proxy_address: str, sell_amount: str, eth_address: str) -> Dict[str, Any]:
+        """
+        Approves the 0x Exchange Proxy contract to spend the specified amount of the ERC-20 token.
+
+        Args:
+        erc20_contract_address (str): The address of the ERC-20 token contract.
+        exchange_proxy_address (str): The address of the 0x Exchange Proxy contract.
+        sell_amount (str): The amount of the token to approve.
+        eth_address (str): Your Ethereum address.
+
+        Returns:
+        dict: The transaction receipt of the approval transaction.
+
+        Example:
+        >>> approve_token('0xTokenAddress', '0xProxyAddress', '1000000000000000000', '0xYourAddress')
+        """
+        erc20_abi = [
+            {
+                "constant": False,
+                "inputs": [
+                    {
+                        "name": "_spender",
+                        "type": "address"
+                    },
+                    {
+                        "name": "_value",
+                        "type": "uint256"
+                    }
+                ],
+                "name": "approve",
+                "outputs": [
+                    {
+                        "name": "",
+                        "type": "bool"
+                    }
+                ],
+                "type": "function"
+            }
+        ]
+        erc20_contract = self.web3.eth.contract(address=erc20_contract_address, abi=erc20_abi)
+        approve_tx = erc20_contract.functions.approve(
+            exchange_proxy_address,
+            int(sell_amount)
+        ).buildTransaction({
+            'from': eth_address,
+            'nonce': self.web3.eth.getTransactionCount(eth_address),
+            'gas': 100000,
+            'gasPrice': self.web3.toWei('20', 'gwei')
+        })
+        signed_approve_tx = self.web3.eth.account.signTransaction(approve_tx, private_key=self.private_key)
+        tx_hash = self.web3.eth.sendRawTransaction(signed_approve_tx.rawTransaction)
+        return self.web3.eth.waitForTransactionReceipt(tx_hash)

src/libs/web3.py

@@ -0,0 +1,8 @@
+import os
+from web3 import Web3
+from dotenv import load_dotenv
+
+load_dotenv()
+
+def get_web3_instance(rpc_ulr: str = os.getenv('INFURA_OR_ALCHEMY_URL')) -> Web3:
+    return Web3(Web3.HTTPProvider(rpc_ulr))

src/llms/sourcegraph.py

@@ -0,0 +1,126 @@
+import os
+from dotenv import load_dotenv
+
+from scrapegraphai.graphs import SearchGraph
+from scrapegraphai.graphs import SmartScraperMultiGraph
+from scrapegraphai.graphs import ScriptCreatorGraph
+
+from src.databases.redis import REDIS_CACHED
+from src.libs.constants import ONE_HOUR_IN_SECONDS
+from src.libs.logger import logger
+
+
+load_dotenv()
+
+redis_cache = REDIS_CACHED
+
+default_graph_config: dict = {
+    "openai": {
+        "llm": {
+            "api_key": os.getenv("OPENAI_API_KEY"),
+            "model": "gpt-3.5-turbo",
+            # "model": "gpt-4o",?
+            "temperature": 0,
+        }
+    },
+    "groq": {
+        "llm": {
+            "model": "groq/llama3-70b-8192",
+            "api_key": os.getenv("GROQ_API_KEY"),
+            "temperature": 0
+        },
+        "embeddings": {
+            "model": "ollama/nomic-embed-text",
+            "base_url": "http://localhost:11434",
+        }
+    }
+}
+
+@redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+@logger.instrument()
+def scrape_graph(prompt: str, sources: str | list[str], config: dict | str = None, cache_ttl: int = None) -> str:
+    """
+    This function scrapes the web using a multi-graph approach. It takes a prompt, sources, optional configuration, and optional cache time-to-live.
+
+    Args:
+        prompt (str): The prompt or query to be used for scraping.
+        sources (str | list[str]): The source(s) from which to scrape data. It can be a single string or a list of strings.
+        config (dict | str, optional): The configuration for the scraping process. Defaults to the default "openai" configuration.
+        cache_ttl (int, optional): The time-to-live (in seconds) for the cached results. Defaults to one hour.
+
+    Returns:
+        str: The scraped data as a string.
+    """
+    logger.info({'prompt': prompt, 'sources': sources, 'config': config, 'cache_ttl': cache_ttl})
+
+    config = type(config) == str and default_graph_config[config] or config or default_graph_config["openai"]
+    logger.debug(f"Config: {config}")
+
+    smart_scraper_graph = SmartScraperMultiGraph(
+        prompt=prompt,
+        source=sources,
+        config=config
+    )
+
+    result = smart_scraper_graph.run()
+    logger.debug(f"Result: {result}")
+
+    return result
+
+@redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+@logger.instrument()
+def search_graph(prompt: str, config: dict | str = None, cache_ttl: int = None) -> str:
+    """
+    This function uses a SearchGraph to search the web for relevant information based on the provided prompt.
+
+    Args:
+        prompt (str): The prompt or query to be used for searching.
+        config (dict | str, optional): The configuration for the search process. Defaults to the default "openai" configuration.
+        cache_ttl (int, optional): The time-to-live (in seconds) for the cached results. Defaults to one hour.
+
+    Returns:
+        str: The search results as a string.
+    """
+    logger.info({'prompt': prompt, 'config': config, 'cache_ttl': cache_ttl})
+
+    config = type(config) == str and default_graph_config[config] or config or default_graph_config["openai"]
+    logger.debug(f"Config: {config}")
+
+    search_graph = SearchGraph(
+        prompt=prompt,
+        config=config,
+    )
+
+    result = search_graph.run()
+    logger.debug(f"Result: {result}")
+
+    return result
+
+@redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+@logger.instrument()
+def _create_script_graph(prompt: str, source: str, library: str, config: dict | str = None, cache_ttl: int = None) -> str:
+    """
+    This function creates a ScriptCreatorGraph for generating scripts based on the provided prompt, source, and library.
+
+    Args:
+        prompt (str): The prompt or query to be used for generating the script.
+        source (str): The source from which to generate the script.
+        library (str): The library to be used for generating the script.
+        config (dict | str, optional): The configuration for the script creation process. Defaults to the default "openai" configuration.
+        cache_ttl (int, optional): The time-to-live (in seconds) for the cached results. Defaults to one hour.
+
+    Returns:
+        str: The generated script as a string.
+    """
+    config = type(config) == str and default_graph_config[config] or config or default_graph_config["openai"]
+
+    script_creator_graph = ScriptCreatorGraph(
+        prompt=prompt,
+        source=source,
+        config=config,
+        library=library
+    )
+
+    result = script_creator_graph.run()
+
+    return result

src/search_services/jina_ai.py

@@ -0,0 +1,100 @@
+from dotenv import load_dotenv
+
+import httpx
+import urllib.parse
+from typing import Self
+from src.libs.helper_functions import convert_to_snakecase
+from src.databases.redis import REDIS_CACHED
+from src.libs.constants import ONE_MINUTE_IN_SECONDS
+from src.libs.constants import JINA_READER_BASE_ENDPOINT, JINA_SEARCH_BASE_ENDPOINT, ONE_MINUTE_IN_SECONDS
+
+load_dotenv()
+
+redis_cache = REDIS_CACHED
+
+class JinaAI:
+    """
+    A class for interacting with Jina AI's search and reader services.
+
+    Attributes:
+        JINA_SEARCH_BASE_ENDPOINT (str): The base URL for the Jina AI search service.
+        JINA_READER_BASE_ENDPOINT (str): The base URL for the Jina AI reader service.
+
+    Methods:
+        __init__(self, search_base_url: str = None, reader_base_url: str = None) -> None:
+            Initialize the JinaAI instance with optional search and reader base URLs.
+
+        search_web_with_jina(self, search_query: str) -> dict | None:
+            Search the web using Jina AI and return the search results as a dictionary.
+
+        read_website_with_jina(self, website_url: str) -> dict | None:
+            Read a website using Jina AI and return the website content as a dictionary.
+    """
+    def __init__(self, search_base_url: str = None, reader_base_url: str = None) -> None:
+        self.JINA_SEARCH_BASE_ENDPOINT = search_base_url or JINA_SEARCH_BASE_ENDPOINT
+        self.JINA_READER_BASE_ENDPOINT = reader_base_url or JINA_READER_BASE_ENDPOINT
+
+    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
+    def search_web_with_jina(self, search_query: str) -> dict | None:
+        """
+        Search the web using Jina AI.
+
+        Args:
+            search_query (str): The query to be searched on the web.
+
+        Returns:
+            dict | None: A dictionary containing the search results if successful, otherwise None.
+
+        Raises:
+            httpx.HTTPError: If an HTTP error occurs during the request.
+
+        Usage:
+            jina_ai_instance.search_web_with_jina(search_query)
+
+        Example:
+            jina_ai_instance.search_web_with_jina("example search query")
+        """
+        url = self.JINA_SEARCH_BASE_ENDPOINT
+        encoded_search_query = urllib.parse.quote(search_query)
+        headers = {"Accept": "application/json"}
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(f"{url}{encoded_search_query}", headers=headers)
+                response.raise_for_status()
+                return convert_to_snakecase(response.json())
+        except httpx.HTTPError as e:
+            print(f"An error occurred: {e}")
+            return None
+        
+    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
+    def read_website_with_jina(self, website_url: str) -> dict | None:
+        """
+        Read a website using Jina AI.
+
+        Args:
+            website_url (str): The URL of the website to be read.
+
+        Returns:
+            dict | None: A dictionary containing the content of the website if successful, otherwise None.
+
+        Raises:
+            httpx.HTTPError: If an HTTP error occurs during the request.
+
+        Usage:
+            jina_ai_instance.read_website_with_jina(website_url)
+
+        Example:
+            jina_ai_instance.read_website_with_jina("https://example.com")
+        """
+        url = self.JINA_READER_BASE_ENDPOINT
+        headers = {"Accept": "application/json"}
+
+        try:
+            with httpx.Client(timeout=30.0) as client:
+                response = client.get(f"{url}{website_url}", headers=headers)
+                response.raise_for_status()
+                return convert_to_snakecase(response.json())
+        except httpx.HTTPError as e:
+            print(f"An error occurred: {e}")
+            return None

src/tools/cloudinary_toolkit.py

@@ -0,0 +1,57 @@
+import asyncio
+import cloudinary
+import cloudinary.uploader
+import cloudinary.api
+from phi.tools import Toolkit
+from phi.utils.log import logger
+
+class CloudinaryTool(Toolkit):
+    def __init__(
+            self,
+            cloud_name: str,
+            api_key: str,
+            api_secret: str,
+    ):
+        super().__init__(name="cloudinary_tool")
+
+        # Initialize Cloudinary
+        cloudinary.config(
+            cloud_name=cloud_name,
+            api_key=api_key,
+            api_secret=api_secret
+        )
+
+        # Registering methods to make them accessible via the toolkit
+        self.register(self.upload_file)
+
+    def upload_file(self, file_path: str) -> str:
+        """
+        Uploads a file to Cloudinary.
+
+        Args:
+        file_path (str): The path to the file to upload.
+
+        Returns:
+        dict: A dictionary containing the upload response details.
+
+        Example:
+        >>> upload_file('/path/to/file.pdf')
+        """
+        logger.info(f"Uploading file to Cloudinary: {file_path}")
+        try:
+            response = cloudinary.uploader.upload(file_path, resource_type="raw")
+            logger.info(f"Upload response: {response}")
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to upload file: {e}")
+            return f"Error: {e}"
+
+# # Example usage
+# cloudinary_tool = CloudinaryTool(
+#     cloud_name="your_cloud_name",
+#     api_key="your_api_key",
+#     api_secret="your_api_secret"
+# )
+#
+# upload_response = cloudinary_tool.upload_file('/path/to/your/file.txt')
+# print(upload_response)

src/tools/crypto_data_toolkit.py

@@ -0,0 +1,151 @@
+from phi.tools import Toolkit
+from phi.utils.log import logger
+
+from src.data_sources.coin_gecko import CoinGecko
+from src.data_sources.cryptocompare import CryptoCompare
+from src.data_sources.dexscreener import DexScreener
+
+class CryptoDataTools(Toolkit):
+    def __init__(self):
+        super().__init__(name="crypto_data_tools")
+        
+        self.register(self.get_coin_price)
+        self.register(self.get_coin_data)
+        self.register(self.get_exchange_data)
+        self.register(self.get_latest_crypto_news_data)
+
+        self.coingecko = CoinGecko()
+        self.crypto_compare = CryptoCompare()
+        self.dex_screener = DexScreener()
+
+    def get_coin_price(self, coin_id: str, vs_currency: str = "usd") -> str:
+        """
+        Fetches the price data for a given cryptocurrency coin from CoinGecko and CryptoCompare.
+
+        Args:
+        coin_id (str): The unique identifier for the cryptocurrency coin.
+        vs_currency (str, optional): The currency to which the coin price will be compared. Defaults to "usd".
+
+        Returns:
+        str: A JSON string containing the price data from CoinGecko and CryptoCompare for the specified coin.
+
+        Raises:
+        Exception: If an error occurs while fetching the price data.
+
+        Example:
+        >>> get_coin_price("bitcoin", "eur")
+        """
+        logger.info(f"Fetching price data for {coin_id} cryptocurrency coin from CoinGecko and CryptoCompare")
+        try:
+            coingecko_price = {}
+            crypto_compare_price = {}
+
+            coingecko_price_data = self.coingecko.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
+            crypto_compare_price_data = self.crypto_compare.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
+
+            logger.debug(f"coingecko_price_data: {coingecko_price_data}")
+            logger.debug(f"crypto_compare_price_data: {crypto_compare_price_data}")
+
+            if coin_id in coingecko_price_data:
+                coingecko_price[coin_id] = coingecko_price_data[coin_id]
+            else:
+                logger.warning(f"Warning: CoinGecko data for {coin_id} not found.")
+
+            if coin_id.upper() in crypto_compare_price_data:
+                crypto_compare_price[coin_id] = crypto_compare_price_data[coin_id.upper()]
+            else:
+                logger.warning(f"Warning: CryptoCompare data for {coin_id} not found.")
+
+            logger.debug(f"response {coingecko_price}")
+            logger.debug(f"response {crypto_compare_price}")
+            
+            return f"{(coingecko_price, crypto_compare_price)}"
+        except Exception as e:
+            logger.warning(f"Failed to fetch price data for {coin_id}: {e}")
+            return f"Error: {e}"
+        
+    def get_coin_data(self, coin_symbol: str) -> str:
+        """
+        Fetches coin data for a given cryptocurrency coin from DexScreener and CryptoCompare.
+
+        Args:
+        coin_symbol (str): The symbol for the cryptocurrency coin.
+
+        Returns:
+        str: A JSON string containing the coin data from DexScreener and CryptoCompare for the specified coin.
+
+        Raises:
+        Exception: If an error occurs while fetching the coin data.
+
+        Example:
+        >>> get_coin_data("BTC")
+        """
+        logger.info(f"Fetching coin data for {coin_symbol} cryptocurrency coin")
+        try:
+            dexscreener_coin_data = self.dex_screener.search(query=coin_symbol)
+            cryptocompare_coin_data = self.crypto_compare.get_overall_coin_data(symbol=coin_symbol)
+
+            logger.debug(f"dexscreener_coin_data: {dexscreener_coin_data}")
+            logger.debug(f"cryptocompare_coin_data: {cryptocompare_coin_data}")
+
+            return f"{(dexscreener_coin_data, cryptocompare_coin_data)}"
+        except Exception as e:
+            logger.warning(f"Failed to fetch coin data for {coin_symbol}: {e}")
+            return f"Error: {e}"
+        
+    def get_exchange_data(self, exchange: str) -> str:
+        """
+        Fetches exchange data for a given cryptocurrency exchange from CoinGecko and DexScreener.
+
+        Args:
+        exchange (str): The unique identifier for the cryptocurrency exchange.
+
+        Returns:
+        str: A JSON string containing the exchange data from CoinGecko and DexScreener for the specified exchange.
+
+        Raises:
+        Exception: If an error occurs while fetching the exchange data.
+
+        Example:
+        >>> get_exchange_data("binance")
+        """
+        logger.info(f"Fetching exchange data for {exchange} cryptocurrency exchange")
+        try:
+            coin_gecko_exchange_data = self.coingecko.get_exchange_data(id=exchange)
+            dexscreener_exchange_data = self.dex_screener.search(query=exchange)
+
+            logger.debug(f"coingecko exchange data: {coin_gecko_exchange_data}")
+            logger.debug(f"dexscreener exchange data: {dexscreener_exchange_data}")
+
+            return f"{(coin_gecko_exchange_data, dexscreener_exchange_data)}"
+        except Exception as e:
+            logger.warning(f"Failed to fetch exchange data for {exchange}: {e}")
+            return f"Error: {e}"
+        
+    def get_latest_crypto_news_data(self) -> str:
+        """
+        Fetches the latest cryptocurrency news data from CoinGecko and CryptoCompare.
+
+        Returns:
+        str: A JSON string containing the latest cryptocurrency news data from CoinGecko and CryptoCompare.
+
+        Raises:
+        Exception: If an error occurs while fetching the latest cryptocurrency news data.
+
+        Example:
+        >>> get_latest_crypto_news_data()
+        """
+        logger.info("Fetching latest cryptocurrency news data")
+        try:
+            coin_gecko_trending_coins = self.coingecko.get_trending_coin_list()
+            cryptocompare_news_categories = self.crypto_compare.get_news_categories()
+            cryptocompare_lates_news_articles = self.crypto_compare.get_latest_news_articles()
+
+            logger.debug(f"coingecko trending coins: {coin_gecko_trending_coins}")
+            logger.debug(f"cryptocompare news categories: {cryptocompare_news_categories}")
+            logger.debug(f"cryptocompare latest news articles: {cryptocompare_lates_news_articles}")
+
+            return f"{(coin_gecko_trending_coins, cryptocompare_news_categories, cryptocompare_lates_news_articles)}"
+        except Exception as e:
+            logger.warning(f"Failed to fetch latest cryptocurrency news data: {e}")
+            return f"Error: {e}"

src/tools/crypto_evm_wallet_toolkit.py

@@ -0,0 +1,97 @@
+import asyncio
+from enum import Enum
+from phi.tools import Toolkit
+from phi.utils.log import logger
+from src.libs.rpc_client import rpc_call
+
+class CryptoEVMWalletTools(Toolkit):
+    def __init__(self):
+        super().__init__(name="crypto_evm_wallet_tools")
+
+        # Registering methods to make them accessible via the toolkit
+        self.register(self.get_evm_wallet_address)
+        self.register(self.get_supported_evm_chains)
+        self.register(self.get_evm_smart_wallet_address)
+        self.chains = self.get_supported_evm_chains()
+
+    def get_supported_evm_chains(self) -> list[str]:
+        """
+        Fetches the list of supported EVM chains.
+
+        Returns:
+        str: A string representation of the response from the RPC call containing the list of supported EVM chains.
+
+        Raises:
+        None
+        """
+        logger.info("Fetching supported EVM chains")
+        
+        params = {}
+        response = asyncio.run(rpc_call(method_name="getEVMSupportedChains", params=params))
+        return f"{response}"
+
+    def get_evm_wallet_address(self, user_email: str, chain: str, testnet: bool = True) -> str:
+        """
+        Fetches an EVM wallet address for the given user email and supported chain.
+        Creates and returns an EVM wallet address for the given user email and supported chain.
+
+        Parameters:
+        - user_email (str): The email of the user for whom the wallet is being created.
+        - chain (ethereum | binance | base | polygon): The EVM chain for which the wallet is being fetched.
+        - testnet (bool, optional): A flag indicating whether the wallet should be on the testnet. Defaults to `True`.
+
+        Returns:
+        - str: A string representation of the response from the RPC call.
+
+        Raises:
+        None
+
+        Note:
+        This method uses asyncio.run() to run the asynchronous RPC call.
+        """
+        logger.info(f"Creating crypto wallet account for {user_email}")
+
+        params = {
+            'chain': chain,
+            'testnet': testnet,
+            'userEmail': user_email,
+        }
+        response = asyncio.run(rpc_call(method_name="getEVMWallet", params=params))
+        return f"{response}"
+    
+    def get_evm_smart_wallet_address(
+        self,
+        user_email: str,
+        chain: str,
+        gasless: bool = True,
+        testnet: bool = True
+    ) -> str:
+        """
+        Fetches a smart EVM wallet address for the given user email and supported chain.
+        Creates and returns a smart EVM wallet address for the given user email and supported chain.
+
+        Parameters:
+        - user_email (str): The email of the user for whom the wallet is being fetched.
+        - chain (ethereum | binance | base | polygon): The EVM chain for which the wallet is being fetched.
+        - gasless (bool, optional): A flag indicating whether the wallet should be gasless. Defaults to `True`.
+        - testnet (bool, optional): A flag indicating whether the wallet should be on the testnet. Defaults to `True`.
+
+        Returns:
+        - str: A string representation of the response from the RPC call.
+
+        Raises:
+        None
+
+        Note:
+        This method uses `asyncio.run()` to run the asynchronous RPC call.
+        """
+        logger.info(f"Fetching crypto wallet account for {user_email}")
+
+        params = {
+            'chain': chain,
+            'gasless': gasless,
+            'testnet': testnet,
+            'userEmail': user_email,
+        }
+        response = asyncio.run(rpc_call(method_name="getEVMSmartWallet", params=params))
+        return f"{response}"

src/tools/crypto_swap_toolkit.py

@@ -0,0 +1,156 @@
+import asyncio
+import json
+import pprint
+import requests
+from web3 import Web3
+from phi.tools import Toolkit
+from phi.utils.log import logger
+from src.libs.web3 import get_web3_instance
+from src.libs.helper_functions import get_headers, get_private_key
+from src.libs.token_approval_helper import TokenApprovalHelper
+from src.libs.rpc_client import rpc_call
+
+class CryptoSwapTools(Toolkit):
+    def __init__(self, web3: Web3 = get_web3_instance()):
+        super().__init__(name="swap_tools")
+
+        # Store Web3 instance
+        self.web3 = web3
+
+        # Helper for token approval
+        self.token_approval_helper = TokenApprovalHelper(web3, get_private_key())
+
+        # Registering methods to make them accessible via the toolkit
+        self.register(self.get_swap_quote)
+        self.register(self.get_swap_price)
+        self.register(self.get_swap_sources)
+        self.register(self.execute_swap)
+
+    def get_swap_quote(self, buy_token: str, sell_token: str, sell_amount: str) -> str:
+        """
+        Fetches a swap quote from the 0x Swap API.
+
+        Args:
+        buy_token (str): The token to buy (e.g., 'DAI').
+        sell_token (str): The token to sell (e.g., 'ETH').
+        sell_amount (str): The amount of the sell token to swap, in the smallest unit (e.g., wei for ETH).
+
+        Returns:
+        dict: A dictionary containing the swap quote details.
+
+        Example:
+        >>> get_swap_quote('DAI', 'ETH', '1000000000000000000')
+        """
+        logger.info(f"Fetching swap quote: buying {buy_token} with {sell_token} amount {sell_amount}")
+        try:
+            params = {
+                'buyToken': buy_token,
+                'sellToken': sell_token,
+                'sellAmount': sell_amount
+            }
+            response = asyncio.run(rpc_call(method_name="getSwapQuote", params=params))
+            return f"{response}"
+        except requests.exceptions.RequestException as e:
+            logger.warning(f"Failed to get swap quote: {e}")
+            # return {"error": str(e)}
+            return f"Error: {e}"
+
+    def get_swap_price(self, buy_token: str, sell_token: str, buy_amount: str) -> str:
+        """
+        Fetches the price for a swap from the 0x Swap API.
+
+        Args:
+        buy_token (str): The token to buy.
+        sell_token (str): The token to sell.
+        buy_amount (str): The amount of the buy token, in the smallest unit.
+
+        Returns:
+        dict: A dictionary containing the swap price details.
+
+        Example:
+        >>> get_swap_price('DAI', 'ETH', '1000000000000000000')
+        """
+        logger.info(f"Fetching swap price: buying {buy_token} with {sell_token} amount {buy_amount}")
+        try:
+            params = {
+                'buyToken': buy_token,
+                'sellToken': sell_token,
+                'buyAmount': buy_amount
+            }
+            response = asyncio.run(rpc_call(method_name="getSwapPrice", params=params))
+            return f"{response}"
+        except requests.exceptions.RequestException as e:
+            logger.warning(f"Failed to get swap price: {e}")
+            # return {"error": str(e)}
+            return f"Error: {e}"
+
+    def get_swap_sources(self) -> str:
+        """
+        Fetches the list of liquidity sources from the 0x Swap API.
+
+        Returns:
+        dict: A dictionary containing the list of liquidity sources.
+
+        Example:
+        >>> get_swap_sources()
+        """
+        logger.info("Fetching swap sources")
+
+        response = asyncio.run(rpc_call(method_name="getSwapSources"))
+        return f"{response}"
+
+    def execute_swap(self, buy_token: str, sell_token: str, sell_amount: str, eth_address: str) -> str:
+        """
+        Executes a swap using the 0x Swap API.
+
+        Args:
+        buy_token (str): The token to buy (e.g., 'DAI').
+        sell_token (str): The token to sell (e.g., 'ETH').
+        sell_amount (str): The amount of the sell token to swap, in the smallest unit (e.g., wei for ETH).
+        eth_address (str): The Ethereum address of the user executing the swap.
+
+        Returns:
+        dict: The transaction receipt of the swap transaction.
+
+        Example:
+        >>> execute_swap('DAI', 'ETH', '1000000000000000000', '0xYourEthereumAddress')
+        """
+        # Get the swap quote
+        quote = json.loads(self.get_swap_quote(buy_token, sell_token, sell_amount))
+        logger.info(f"Swap quote: {quote}")
+
+        if 'error' in quote:
+            # return {"error": "Failed to get swap quote"}
+            return f"Error: Failed to get swap quote"
+
+        # Approve the token if needed (skip if selling ETH)
+        if sell_token != 'ETH':
+            approval_receipt = self.token_approval_helper.approve_token(sell_token, quote['allowanceTarget'],
+                                                                        sell_amount, eth_address)
+            logger.info(f"Approval receipt: {approval_receipt}")
+
+            if 'status' not in approval_receipt or approval_receipt['status'] != 1:
+                # return {"error": "Token approval failed"}
+                return f"Error: Token approval failed"
+
+        # Execute the swap
+        try:
+            swap_tx = {
+                'from': eth_address,
+                'to': quote['to'],
+                'data': quote['data'],
+                'value': int(quote['value']),
+                'gas': 200000,
+                'gasPrice': self.web3.to_wei('20', 'gwei'),
+                'nonce': self.web3.eth.get_transaction_count(eth_address)
+            }
+            signed_swap_tx = self.web3.eth.account.signTransaction(swap_tx, private_key=get_private_key())
+            tx_hash = self.web3.eth.send_raw_transaction(signed_swap_tx.rawTransaction)
+            receipt = self.web3.eth.wait_for_transaction_receipt(tx_hash)
+            logger.info(f"Swap transaction receipt: {receipt}")
+            # return receipt
+            return f"{(receipt)}"
+        except Exception as e:
+            logger.warning(f"Failed to execute swap: {e}")
+            # return {"error": str(e)}
+            return f"Error: {e}"

src/tools/notepad_toolkit.py

@@ -0,0 +1,214 @@
+import asyncio
+import chainlit as cl
+from phi.tools import Toolkit
+from phi.utils.log import logger
+from src.libs.rpc_client import rpc_call  # Replace with the actual path to your RPC client library
+
+class NotedPadToolkit(Toolkit):
+    def __init__(self):
+        super().__init__(name="noted_pad_toolkit")
+
+        # Registering methods to make them accessible via the toolkit
+        self.register(self.get_note)
+        self.register(self.get_all_notes)
+        self.register(self.add_note)
+        self.register(self.update_note)
+        self.register(self.delete_note)
+        self.register(self.query_notes)
+
+    def get_user_email(self) -> str:
+        """
+        Fetches the user's email from the Chainlit user session.
+
+        Returns:
+        str: The user's email.
+
+        Example:
+        >>> get_user_email()
+        """
+        logger.info("Fetching user email")
+        try:
+            user_session = cl.user_session.get("user").metadata
+            if not user_session:
+                raise ValueError("User session not found")
+            email = user_session['email']
+            return email
+        except Exception as e:
+            logger.warning(f"Failed to get user email: {e}")
+            return f"Error: {e}"
+
+    @cl.on_chat_start
+    def get_note(self, note_id: str) -> str:
+        """
+        Fetches a specific note from the RPC server.
+
+        Args:
+        note_id (str): The ID of the note to fetch.
+
+        Returns:
+        str: The requested note.
+
+        Example:
+        >>> get_note('note_1')
+        """
+        logger.info(f"Fetching note: {note_id}")
+        try:
+            email = self.get_user_email()
+            if "Error" in email:
+                raise ValueError(email)
+
+            params = {
+                'email': email,
+                'note_id': note_id
+            }
+            response = asyncio.run(rpc_call(method_name="getNote", params=params))
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to get note: {e}")
+            return f"Error: {e}"
+
+    def get_all_notes(self) -> str:
+        """
+        Fetches all notes from the RPC server.
+
+        Returns:
+        dict: A dictionary containing all notes.
+
+        Example:
+        >>> get_all_notes()
+        """
+        logger.info("Fetching all notes")
+        try:
+            email = self.get_user_email()
+            if "Error" in email:
+                raise ValueError(email)
+
+            params = {
+                'email': email
+            }
+            print(params)
+            response = asyncio.run(rpc_call(method_name="getAllNotes", params=params))
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to get all notes: {e}")
+            return f"Error: {e}"
+
+    def add_note(self, note_id: str, note_content: str) -> str:
+        """
+        Adds a new note to the RPC server. add important notes during conversationß
+
+        Args:
+        note_id (str): The ID of the note to add.
+        note_content (str): The content of the note.
+
+        Returns:
+        str: Confirmation message.
+
+        Example:
+        >>> add_note('note_1', 'This is a new note')
+        """
+        logger.info(f"Adding note: {note_id}")
+        try:
+            email = self.get_user_email()
+            if "Error" in email:
+                raise ValueError(email)
+
+            params = {
+                'email': email,
+                'note_id': note_id,
+                'note_content': note_content
+            }
+            response = asyncio.run(rpc_call(method_name="addNote", params=params))
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to add note: {e}")
+            return f"Error: {e}"
+
+    def update_note(self, note_id: str, note_content: str) -> str:
+        """
+        Updates an existing note on the RPC server.
+
+        Args:
+        note_id (str): The ID of the note to update.
+        note_content (str): The new content of the note.
+
+        Returns:
+        str: Confirmation message.
+
+        Example:
+        >>> update_note('note_1', 'This is an updated note')
+        """
+        logger.info(f"Updating note: {note_id}")
+        try:
+            email = self.get_user_email()
+            if "Error" in email:
+                raise ValueError(email)
+
+            params = {
+                'email': email,
+                'note_id': note_id,
+                'note_content': note_content
+            }
+            response = asyncio.run(rpc_call(method_name="updateNote", params=params))
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to update note: {e}")
+            return f"Error: {e}"
+
+    def delete_note(self, note_id: str) -> str:
+        """
+        Deletes a note from the RPC server.
+
+        Args:
+        note_id (str): The ID of the note to delete.
+
+        Returns:
+        str: Confirmation message.
+
+        Example:
+        >>> delete_note('note_1')
+        """
+        logger.info(f"Deleting note: {note_id}")
+        try:
+            email = self.get_user_email()
+            if "Error" in email:
+                raise ValueError(email)
+
+            params = {
+                'email': email,
+                'note_id': note_id
+            }
+            response = asyncio.run(rpc_call(method_name="deleteNote", params=params))
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to delete note: {e}")
+            return f"Error: {e}"
+
+    def query_notes(self, query: str) -> str:
+        """
+        Queries notes from the RPC server based on the provided context.
+
+        Args:
+        query (str): The context or keywords to search for in notes.
+
+        Returns:
+        str: The notes that match the query.
+
+        Example:
+        >>> query_notes('spiritual growth')
+        """
+        logger.info(f"Querying notes with context: {query}")
+        try:
+            email = self.get_user_email()
+            if "Error" in email:
+                raise ValueError(email)
+
+            params = {
+                'email': email,
+                'query': query
+            }
+            response = asyncio.run(rpc_call(method_name="queryNotes", params=params))
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to query notes: {e}")
+            return f"Error: {e}"

src/tools/user_profile_toolkit.py

@@ -0,0 +1,185 @@
+import chainlit as cl
+from phi.tools import Toolkit
+from phi.utils.log import logger
+
+
+class UserProfileToolkit(Toolkit):
+    def __init__(self):
+        super().__init__(name="user_profile_toolkit")
+
+        # Registering methods to make them accessible via the toolkit
+        self.register(self.get_user_name)
+        self.register(self.get_user_email)
+        self.register(self.get_user_picture)
+        self.register(self.get_all_user_info)
+        self.register(self.update_user_info)
+        self.register(self.update_user_name)
+        self.register(self.update_user_email)
+        self.register(self.update_user_picture)
+
+    @cl.on_chat_start
+    def get_user_info(self, info_type: str) -> str:
+        """
+        Fetches user information from the Chainlit user session.
+
+        Args:
+        info_type (str): The type of information to fetch ('name', 'email', 'picture').
+
+        Returns:
+        str: The requested user information.
+
+        Example:
+        >>> get_user_info('name')
+        """
+        logger.info(f"Fetching user info: {info_type}")
+        try:
+            # Fetch user session data
+            user_session = cl.user_session.get("user").metadata
+            if not user_session:
+                raise ValueError("User session not found")
+
+            response = user_session[info_type]
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to get user info: {e}")
+            return f"Error: {e}"
+
+    def get_all_user_info(self) -> str:
+        """
+        Fetches all user information from the Chainlit user session.
+
+        Returns:
+        dict: A dictionary containing all user information.
+
+        Example:
+        >>> get_all_user_info()
+        """
+        logger.info("Fetching all user info")
+        try:
+            # Fetch user session data
+            user_session = cl.user_session.get("user").metadata
+            if not user_session:
+                raise ValueError("User session not found")
+
+            response = user_session
+            return f"{response}"
+        except Exception as e:
+            logger.warning(f"Failed to get all user info: {e}")
+            # return {"error": str(e)}
+            return f"Error: {e}"
+
+    def update_user_info(self, info_type: str, value: str) -> str:
+        """
+        Updates user information in the Chainlit user session.
+
+        Args:
+        info_type (str): The type of information to update ('name', 'email', 'picture').
+        value (str): The new value to set.
+
+        Returns:
+        str: Confirmation message.
+
+        Example:
+        >>> update_user_info('name', 'Jane Doe')
+        """
+        logger.info(f"Updating user info: {info_type} to {value}")
+        try:
+            # Fetch user session data
+            user_session = cl.user_session.get("user").metadata
+            if not user_session:
+                raise ValueError("User session not found")
+
+            # Update the user session data
+            user_session[info_type] = value
+            cl.user_session.set("user", user_session)
+            return f"{info_type} updated to {value}"
+        except Exception as e:
+            logger.warning(f"Failed to update user info: {e}")
+            return f"Error: {e}"
+
+    def get_user_name(self) -> str:
+        """
+        Fetches the user's name from the Chainlit user session.
+
+        Returns:
+        str: The user's name.
+
+        Example:
+        >>> get_user_name()
+        """
+        response = self.get_user_info('name')
+        return f"{response}"
+
+    def get_user_email(self) -> str:
+        """
+        Fetches the user's email from the Chainlit user session.
+
+        Returns:
+        str: The user's email.
+
+        Example:
+        >>> get_user_email()
+        """
+        response = self.get_user_info('email')
+        return f"{response}"
+
+    def get_user_picture(self) -> str:
+        """
+        Fetches the user's picture URL from the Chainlit user session.
+
+        Returns:
+        str: The user's picture URL.
+
+        Example:
+        >>> get_user_picture()
+        """
+        response = self.get_user_info('picture')
+        return f"{response}"
+
+    def update_user_name(self, name: str) -> str:
+        """
+        Updates the user's name in the Chainlit user session.
+
+        Args:
+        name (str): The new name to set.
+
+        Returns:
+        str: Confirmation message.
+
+        Example:
+        >>> update_user_name('Jane Doe')
+        """
+        response = self.update_user_info('name', name)
+        return f"{response}"
+
+    def update_user_email(self, email: str) -> str:
+        """
+        Updates the user's email in the Chainlit user session.
+
+        Args:
+        email (str): The new email to set.
+
+        Returns:
+        str: Confirmation message.
+
+        Example:
+        >>> update_user_email('jane.doe@example.com')
+        """
+        response = self.update_user_info('email', email)
+        return f"{response}"
+
+    def update_user_picture(self, picture: str) -> str:
+        """
+        Updates the user's picture URL in the Chainlit user session.
+
+        Args:
+        picture (str): The new picture URL to set.
+
+        Returns:
+        str: Confirmation message.
+
+        Example:
+        >>> update_user_picture('https://example.com/new_picture.jpg')
+        """
+        response = self.update_user_info('picture', picture)
+        return f"{response}"