Enhancement: Add RSI and MACD technical indicators #11

app.py

@@ -1,13 +1,18 @@
 import streamlit as st
-import pandas as pd
 from src.data.b3_data import B3Data
-from src.utils.investment_calculator import calculate_investment_value, investment_growth_schedule
+from src.utils.investment_calculator import (
+    calculate_investment_value, investment_growth_schedule
+)
+
 
 def main():
     st.title("Brazilian Stocks Dashboard")
 
     # Input for tickers
-    tickers_input = st.text_input("Enter stock tickers (comma-separated, e.g., PETR4.SA,VALE3.SA)", "PETR4.SA,VALE3.SA")
+    tickers_input = st.text_input(
+        "Enter stock tickers (comma-separated, e.g., PETR4.SA,VALE3.SA)",
+        "PETR4.SA,VALE3.SA"
+    )
 
     if tickers_input:
         ticker_list = [ticker.strip() for ticker in tickers_input.split(',')]
@@ -21,48 +26,69 @@ def main():
             st.dataframe(stock_data)
         else:
             st.warning("No data found for the given tickers.")
-    
+
     # Investment Calculator Section
     st.markdown("---")
     st.subheader("💰 Investment Calculator")
-    
+
     col1, col2 = st.columns(2)
-    
+
     with col1:
-        principal = st.number_input("Initial Investment (R$)", min_value=0.0, value=10000.0, step=100.0)
-        monthly_contribution = st.number_input("Monthly Contribution (R$)", min_value=0.0, value=500.0, step=50.0)
-
+        principal = st.number_input(
+            "Initial Investment (R$)", min_value=0.0, value=10000.0, step=100.0
+        )
+        monthly_contribution = st.number_input(
+            "Monthly Contribution (R$)", min_value=0.0, value=500.0, step=50.0
+        )
+
     with col2:
-        annual_rate = st.number_input("Annual Interest Rate (%)", min_value=0.0, max_value=100.0, value=7.0, step=0.1)
-        years = st.number_input("Investment Period (years)", min_value=1, max_value=50, value=10, step=1)
-
+        annual_rate = st.number_input(
+            "Annual Interest Rate (%)",
+            min_value=0.0, max_value=100.0, value=7.0, step=0.1
+        )
+        years = st.number_input(
+            "Investment Period (years)",
+            min_value=1, max_value=50, value=10, step=1
+        )
+
     # Convert percentage to decimal
     annual_rate_decimal = annual_rate / 100
-    
+
     if st.button("Calculate Investment"):
-        final_value = calculate_investment_value(principal, monthly_contribution, annual_rate_decimal, years)
-
+        final_value = calculate_investment_value(
+            principal, monthly_contribution, annual_rate_decimal, years
+        )
+
         col1, col2, col3 = st.columns(3)
-        
+
         with col1:
-            st.metric("Final Value", f"R$ {final_value:,.2f}")
-        
+            st.metric("Final Value", f"R$ {final_value: ,.2f}")
+
         with col2:
-            total_contributions = principal + (monthly_contribution * 12 * years)
-            st.metric("Total Contributions", f"R$ {total_contributions:,.2f}")
-
+            total_contributions = principal + \
+                (monthly_contribution * 12 * years)
+            st.metric(
+                "Total Contributions", f"R$ {total_contributions: ,.2f}"
+            )
+
         with col3:
             total_interest = final_value - total_contributions
-            st.metric("Interest Earned", f"R$ {total_interest:,.2f}")
-        
+            st.metric("Interest Earned", f"R$ {total_interest: ,.2f}")
+
         # Growth schedule
         st.subheader("Investment Growth Schedule")
-        schedule_df = investment_growth_schedule(principal, monthly_contribution, annual_rate_decimal, years)
+        schedule_df = investment_growth_schedule(
+            principal, monthly_contribution, annual_rate_decimal, years
+        )
         st.dataframe(schedule_df)
-        
+
         # Chart
         st.subheader("Investment Growth Chart")
-        st.line_chart(schedule_df.set_index('year')[['balance', 'total_contributions']])
+        st.line_chart(
+            schedule_df.set_index('year')
+            [['balance', 'total_contributions']]
+        )
+
 
 if __name__ == "__main__":
     main()

examples/fetch_b3.py

@@ -1,2 +1,2 @@
 from src.data import get_stock_data
-print(get_stock_data(["PETR4","VALE3"], period="1mo").head())
+print(get_stock_data(["PETR4", "VALE3"], period="1mo").head())

requirement.txt

@@ -1,3 +1,4 @@
 pandas>=2.0.0
 yfinance>=0.2.40
 numpy>=1.24.0
+pandas-ta>=0.3.14b

src/data/__init__.py

@@ -1,3 +1,3 @@
-from .b3_data import B3Data, get_stock_data
+from .b3_data import B3Data
 
-__all__ = ["B3Data", "get_stock_data"]
+__all__ = ["B3Data"]

src/data/b3_data.py

@@ -1,90 +1,97 @@
-from __future__ import annotations
-from typing import Iterable, List, Union
-import pandas as pd
 import yfinance as yf
+import pandas as pd
+from typing import List, Optional
 
-_COLMAP = {"Open": "open", "High": "high", "Low": "low", "Close": "close", "Volume": "volume"}
 
 class B3Data:
-    def _norm(self, t: str) -> str:
-        t = t.strip().upper()
-        return t if t.endswith(".SA") else f"{t}.SA"
+    """
+    A class to fetch and process B3 stock data using yfinance.
+    """
+
+    def __init__(
+        self,
+        start_date: Optional[str] = None,
+        end_date: Optional[str] = None
+    ):
+        """
+        Initializes the B3Data class.
+        Args:
+            start_date (str, optional): The start date for fetching data
+                in 'YYYY-MM-DD' format. Defaults to None.
+            end_date (str, optional): The end date for fetching data in
+                'YYYY-MM-DD' format. Defaults to None.
+        """
+        self.start_date = start_date
+        self.end_date = end_date
 
     def get_stock_data(
         self,
-        ticker_list: Union[str, Iterable[str]],
+        tickers: List[str],
         period: str = "1y",
-        interval: str = "1d",
-        auto_adjust: bool = True,
+        interval: str = "1d"
     ) -> pd.DataFrame:
         """
-        Return OHLCV for B3 tickers using yfinance.
-        Columns: ['ticker','date','open','high','low','close','volume']
+        Fetches historical data for a list of stock tickers.
+        Args:
+            tickers (List[str]): A list of stock tickers
+                (e.g., ['PETR4.SA', 'VALE3.SA']).
+            period (str, optional): The period to fetch data for (e.g., '1d',
+                '5d', '1mo', '3mo', '6mo', '1y', '2y', '5y', '10y', 'ytd',
+                'max'). Defaults to "1y".
+            interval (str, optional): The data interval (e.g., '1m', '2m',
+                '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1wk',
+                '1mo', '3mo'). Defaults to "1d".
+        Returns:
+            pd.DataFrame: A DataFrame containing the historical data for the
+                given tickers.
         """
-        tickers = [ticker_list] if isinstance(ticker_list, str) else list(ticker_list)
-        if not tickers:
-            raise ValueError("ticker_list must not be empty")
-        tickers = [self._norm(t) for t in tickers]
+        if not isinstance(tickers, list) or not all(
+            isinstance(ticker, str) for ticker in tickers
+        ):
+            raise ValueError("Tickers must be a list of strings.")
 
-        frames: List[pd.DataFrame] = []
-        for t in tickers:
-            hist = yf.Ticker(t).history(period=period, interval=interval, auto_adjust=auto_adjust)
-            if hist.empty:
-                continue
-            hist = hist.rename(columns=_COLMAP)
-            keep = [c for c in ["open", "high", "low", "close", "volume"] if c in hist.columns]
-            hist = hist[keep]
-            hist["ticker"] = t
-            hist = hist.reset_index().rename(columns={"Date": "date"})
-            frames.append(hist)
+        try:
+            data = yf.download(
+                tickers,
+                period=period,
+                interval=interval,
+                start=self.start_date,
+                end=self.end_date
+            )
+            if data.empty:
+                return pd.DataFrame()
 
-        if not frames:
-            return pd.DataFrame(columns=["ticker", "date", "open", "high", "low", "close", "volume"])
+            # Restructure dataframe for better readability
+            data.columns = data.columns.swaplevel(0, 1)
+            data.sort_index(axis=1, level=0, inplace=True)
 
-        df = pd.concat(frames, ignore_index=True)
-        return df.sort_values(["ticker", "date"]).reset_index(drop=True)
+            return data
+        except Exception as e:
+            print(f"An error occurred: {e}")
+            return pd.DataFrame()
 
-    def __init__(self):
-        pass
-
-    def get_stock_data(self, tickers, period="1mo"):
+    def calculate_technical_indicators(
+        self, data: pd.DataFrame
+    ) -> pd.DataFrame:
         """
-        Fetch stock data for given tickers from Yahoo Finance
-
+        Calculates technical indicators for the given stock data.
         Args:
-            tickers (list): List of stock tickers
-            period (str): Period for data (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)
-
+            data (pd.DataFrame): A DataFrame with stock data,
+                multi-indexed by ticker.
         Returns:
-            pandas.DataFrame: Stock data
+            pd.DataFrame: A DataFrame with the technical indicators.
         """
-        try:
-            data = yf.download(tickers, period=period, group_by='ticker')
-
-            if len(tickers) == 1:
-                # Single ticker case
-                df = data.reset_index()
-                df['Ticker'] = tickers[0]
-                return df
-            else:
-                # Multiple tickers case
-                combined_data = []
-                for ticker in tickers:
-                    if ticker in data.columns.levels[0]:
-                        ticker_data = data[ticker].reset_index()
-                        ticker_data['Ticker'] = ticker
-                        combined_data.append(ticker_data)
-
-                if combined_data:
-                    return pd.concat(combined_data, ignore_index=True)
-                else:
-                    return pd.DataFrame()
-
-        except Exception as e:
-            print(f"Error fetching data: {e}")
+        if data.empty:
             return pd.DataFrame()
 
-def get_stock_data(tickers, period="1mo"):
-    """Standalone function for backwards compatibility"""
-    b3_data = B3Data()
-    return b3_data.get_stock_data(tickers, period)
+        df_with_indicators = data.copy()
+
+        # Example: Calculate SMA for each stock
+        for ticker in df_with_indicators.columns.levels[0]:
+            df_with_indicators[(ticker, 'SMA_20')] = (
+                df_with_indicators[(ticker, 'Close')]
+                .rolling(window=20)
+                .mean()
+            )
+
+        return df_with_indicators

src/data/fx_scraper.py

@@ -1,6 +1,7 @@
 import requests
 from bs4 import BeautifulSoup
 
+
 def get_usd_brl_rate():
     """
     Scrapes USD to BRL exchange rate from a public source.
@@ -12,12 +13,12 @@ def get_usd_brl_rate():
         raise Exception("Failed to fetch FX rate page")
 
     soup = BeautifulSoup(response.text, "html.parser")
-    
+
     # Find the rate element (site-specific selector)
     rate_tag = soup.find("span", class_="ccOutputRslt")
     if not rate_tag:
         raise Exception("Exchange rate not found in page")
-    
+
     # Clean and convert to float
     rate_text = rate_tag.get_text().split(" ")[0]
     return float(rate_text.replace(",", ""))

src/ml/predict.py

@@ -3,6 +3,7 @@
 def predict_price(ticker):
     """
     Placeholder function to predict stock price.
-    Can be implemented later with a machine learning model (e.g., Linear Regression).
+    Can be implemented later with a machine learning model
+    (e.g., Linear Regression).
     """
     pass  # To be implemented later

src/utils/__init__.py

@@ -1 +0,0 @@
-from .investment_calculator import calculate_investment_value, investment_growth_schedule