Update algo_trade_poc.py

algo_trade_poc.py

@@ -1,189 +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__":
+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",progress=False)
+        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()