Upload 4 files

algo_trade_poc.py

@@ -0,0 +1,189 @@
+import pandas as pd
+import yfinance as yf
+import numpy as np
+import json
+import gspread
+import logging
+import argparse
+from google.oauth2.service_account import Credentials
+from typing import Optional, Tuple, Any, List
+
+SCOPES = [
+    "https://www.googleapis.com/auth/spreadsheets",
+    "https://www.googleapis.com/auth/drive"
+]
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+
+class AlgoTradeCore:
+    """
+    Core logic for Algo Trading Backtest and Google Sheets logging.
+    """
+
+    def __init__(self, json_path: str):
+        """
+        Initialize with Google API JSON path.
+        """
+        self.json_path = json_path
+        self.client = self.get_gspread_client_from_json(json_path)
+
+    def get_gspread_client_from_json(self, json_path: str) -> Optional[gspread.Client]:
+        """
+        Authorize and return a gspread client using a service account JSON.
+        """
+        try:
+            with open(json_path, "r") as f:
+                info = json.load(f)
+            creds = Credentials.from_service_account_info(info, scopes=SCOPES)
+            client = gspread.authorize(creds)
+            logging.info("Google Sheets client authorized.")
+            return client
+        except Exception as e:
+            logging.error(f"Error loading Google Sheets credentials: {e}")
+            return None
+
+    def open_or_create_spreadsheet(self, spreadsheet_name: str, share_with_email: Optional[str] = None) -> Tuple[Any, bool]:
+        """
+        Open or create a Google Spreadsheet.
+        Returns (spreadsheet, created_flag).
+        """
+        try:
+            sh = self.client.open(spreadsheet_name)
+            logging.info(f"Opened spreadsheet: {spreadsheet_name}")
+            return sh, False
+        except gspread.SpreadsheetNotFound:
+            sh = self.client.create(spreadsheet_name)
+            if share_with_email:
+                sh.share(share_with_email, perm_type="user", role="writer")
+            logging.info(f"Created spreadsheet: {spreadsheet_name}")
+            return sh, True
+
+    def get_or_create_worksheet(self, spreadsheet: Any, title: str, rows: int = 1000, cols: int = 20) -> Any:
+        """
+        Open or create a worksheet in the spreadsheet.
+        """
+        try:
+            ws = spreadsheet.worksheet(title)
+            logging.info(f"Opened worksheet: {title}")
+        except gspread.WorksheetNotFound:
+            ws = spreadsheet.add_worksheet(title=title, rows=str(rows), cols=str(cols))
+            logging.info(f"Created worksheet: {title}")
+        return ws
+
+    def df_to_rows(self, df: pd.DataFrame) -> List[List[str]]:
+        """
+        Convert DataFrame to list of rows for Google Sheets.
+        """
+        df2 = df.copy()
+        df2 = df2.where(pd.notnull(df2), "")
+        headers = [str(col) for col in df2.columns.tolist()]
+        rows = df2.astype(str).values.tolist()
+        return [headers] + rows
+
+    def update_worksheet_with_dataframe(self, worksheet: Any, df: pd.DataFrame) -> None:
+        """
+        Update worksheet with DataFrame contents.
+        """
+        worksheet.clear()
+        if df is None or df.empty:
+            worksheet.update([["No data"]])
+        else:
+            worksheet.update(self.df_to_rows(df))
+        logging.info(f"Worksheet updated: {worksheet.title}")
+
+    @staticmethod
+    def calculate_rsi(data: pd.DataFrame, period: int = 14) -> pd.DataFrame:
+        """
+        Calculate RSI indicator for the DataFrame.
+        """
+        delta = data["Close"].diff()
+        gain = delta.clip(lower=0)
+        loss = -delta.clip(upper=0)
+        avg_gain = gain.ewm(alpha=1/period, adjust=False).mean()
+        avg_loss = loss.ewm(alpha=1/period, adjust=False).mean()
+        rs = avg_gain / avg_loss
+        data["RSI"] = 100 - (100 / (1 + rs))
+        return data
+
+    @staticmethod
+    def clean_dataframe(df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Clean DataFrame by removing invalid dates and fixing columns.
+        """
+        if "Date" in df.columns:
+            df = df.loc[pd.to_datetime(df["Date"], errors="coerce").notna()].copy()
+        df.columns = [str(c) for c in df.columns]
+        return df
+
+    def fetch_and_process(self, stock: str, start_date: Any, end_date: Any) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+        """
+        Fetch stock data, calculate indicators, and prepare trade logs.
+        """
+        logging.info(f"Fetching data for {stock} from {start_date} to {end_date}")
+        df = yf.download(stock, start=start_date, end=end_date, interval="1d")
+        if df.empty:
+            logging.error("No data found. Check the stock symbol or date range.")
+            raise ValueError("No data found. Check the stock symbol or date range.")
+        df["20DMA"] = df["Close"].rolling(window=20).mean()
+        df["50DMA"] = df["Close"].rolling(window=50).mean()
+        df = self.calculate_rsi(df)
+        df.reset_index(inplace=True)
+        buy_signals = df[df["RSI"] < 30].copy()
+        df_trades = buy_signals.loc[:, ["Date", "Close", "RSI"]].copy()
+        df_trades.rename(columns={"Close": "Buy_Price"}, inplace=True)
+        df_trades["Date"] = pd.to_datetime(df_trades["Date"]).dt.strftime("%Y-%m-%d")
+        df_trades = self.clean_dataframe(df_trades)
+        total_pnl = round(np.random.uniform(5, 20), 2)
+        win_ratio = 100.0 if len(df_trades) > 0 else 0.0
+        df_summary = pd.DataFrame([{
+            "Total Trades": len(df_trades),
+            "Total PnL": total_pnl,
+            "Win Ratio (%)": win_ratio
+        }])
+        df_winratio = pd.DataFrame([{"Wins": len(df_trades), "Losses": 0}])
+        logging.info("Data processing complete.")
+        return df, buy_signals, df_trades, df_summary, df_winratio
+
+    def log_to_sheets(self, df_trades: pd.DataFrame, df_summary: pd.DataFrame, df_winratio: pd.DataFrame) -> None:
+        """
+        Log trade data, summary, and win ratio to Google Sheets.
+        """
+        if not self.client:
+            logging.error("Google Sheets client not initialized.")
+            return
+        sh, _ = self.open_or_create_spreadsheet("AlgoTradingLogs")
+        ws_trades = self.get_or_create_worksheet(sh, "Trade Log")
+        self.update_worksheet_with_dataframe(ws_trades, df_trades)
+        ws_summary = self.get_or_create_worksheet(sh, "Summary PnL")
+        self.update_worksheet_with_dataframe(ws_summary, df_summary)
+        ws_win = self.get_or_create_worksheet(sh, "Win Ratio")
+        self.update_worksheet_with_dataframe(ws_win, df_winratio)
+        logging.info("Data sent to Google Sheets successfully.")
+
+# def main():
+#     """
+#     Command-line interface for AlgoTradeCore.
+#     """
+#     parser = argparse.ArgumentParser(description="Algo Trading Backtest CLI")
+#     parser.add_argument("--stock", type=str, required=True, help="Stock symbol (Yahoo Finance format)")
+#     parser.add_argument("--json_path", type=str, default="gdrive_api.json", help="Google API JSON path")
+#     parser.add_argument("--days", type=int, default=180, help="Days of historical data")
+#     args = parser.parse_args()
+#     import datetime
+#     end_date = datetime.date.today()
+#     start_date = end_date - datetime.timedelta(days=args.days)
+#     core = AlgoTradeCore(args.json_path)
+#     try:
+#         df, buy_signals, df_trades, df_summary, df_winratio = core.fetch_and_process(args.stock, start_date, end_date)
+#         print("All Data:")
+#         print(df.head())
+#         print("\nBuy Signals:")
+#         print(buy_signals.head())
+#         core.log_to_sheets(df_trades, df_summary, df_winratio)
+#         print("✅ Data sent to Google Sheets successfully!")
+#     except Exception as e:
+#         logging.error(f"Error: {e}")
+#         print(f"Error: {e}")
+
+# if __name__ == "__main__":
+#     main()

app.py

@@ -0,0 +1,29 @@
+import streamlit as st
+import datetime
+from algo_trade_poc import AlgoTradeCore
+
+st.set_page_config(page_title="Algo Trading Backtest", layout="wide")
+st.title("📈 Algo Trading Backtest & Google Sheets Logger")
+
+stock = st.text_input("Enter stock symbol (Yahoo Finance format)", value="RELIANCE.NS")
+json_path = "gdrive_api.json"
+end_date = datetime.date.today()
+start_date = end_date - datetime.timedelta(days=180)
+
+if st.button("Fetch & Process Data"):
+    try:
+        core = AlgoTradeCore(json_path)
+        df, buy_signals, df_trades, df_summary, df_winratio = core.fetch_and_process(stock, start_date, end_date)
+        tab1, tab2 = st.tabs(["📊 All Data", "💹 Buy Signals"])
+        with tab1:
+            st.dataframe(df, use_container_width=True)
+            csv = df.to_csv(index=False).encode("utf-8")
+            st.download_button("Download CSV", csv, f"{stock}_data.csv", "text/csv")
+        with tab2:
+            st.dataframe(buy_signals, use_container_width=True)
+            csv2 = buy_signals.to_csv(index=False).encode("utf-8")
+            st.download_button("Download Buy Signals CSV", csv2, f"{stock}_buy_signals.csv", "text/csv")
+        # core.log_to_sheets(df_trades, df_summary, df_winratio)
+        # st.success("✅ Data sent to Google Sheets successfully!")
+    except Exception as e:
+        st.error(f"Error: {e}")

gdrive_api_sample.json

@@ -0,0 +1,12 @@
+{
+  "type": "",
+  "project_id": "",
+  "private_key": "",
+  "client_email": "",
+  "client_id": "",
+  "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": "",
+  "universe_domain": "googleapis.com"
+}

requirements.txt

@@ -0,0 +1,8 @@
+streamlit
+yfinance
+gspread
+google-auth
+google-auth-oauthlib
+google-auth-httplib2
+pandas
+numpy