Implement Weather API for better trade improvements (#13)

Related to #3

Add Weather API integration to improve trading strategies.

* **Configuration Changes:**
- Add `WEATHER_API_KEY` and `WEATHER_API_ENDPOINT` to
`config/api_config.py` and `.env`.
  - Add weather API configurations to `config/config.py`.

* **Weather Data Fetching:**
- Create `data/weather_data_fetcher.py` to fetch weather data from the
Weather API.
- Initialize `WeatherDataFetcher` in `data/data_fetcher.py` and add
method `fetch_and_process_weather_data`.

* **Strategy Updates:**
- Modify `strategies/futures_strategy.py` and
`strategies/options_strategy.py` to incorporate weather data in trading
decisions.
- Add methods to analyze weather data and generate signals based on
weather conditions.

* **Main Function Updates:**
  - Import and initialize `WeatherDataFetcher` in `main.py`.
- Modify `process_market_data` function to include weather data
processing and notifications.

.env

@@ -30,3 +30,7 @@ LOG_LEVEL=INFO
 # Telegram Notifications
 TELEGRAM_BOT_API_KEY=your_telegram_bot_api_key
 TELEGRAM_CHAT_ID=your_telegram_chat_id
+
+# Weather API
+WEATHER_API_KEY=your_weather_api_key
+WEATHER_API_ENDPOINT=your_weather_api_endpoint

config/api_config.py

@@ -22,6 +22,10 @@ def __init__(self):
         self.chatgpt_api_key = os.getenv("CHATGPT_API_KEY")
         self.chatgpt_endpoint = "https://api.openai.com/v1/chat/completions"  # Endpoint for ChatGPT API
 
+        # Weather API configurations
+        self.weather_api_key = os.getenv("WEATHER_API_KEY")
+        self.weather_api_endpoint = os.getenv("WEATHER_API_ENDPOINT")
+
         # Additional API configurations can be added here
         # Example for other market data or sentiment analysis APIs:
         # self.other_api_key = os.getenv("OTHER_API_KEY")
@@ -45,6 +49,15 @@ def chatgpt_headers(self):
             "Content-Type": "application/json",
         }
 
+    def weather_headers(self):
+        """
+        Returns headers required for making requests to the Weather API.
+        """
+        return {
+            "Authorization": f"Bearer {self.weather_api_key}",
+            "Content-Type": "application/json",
+        }
+
     def get_dhan_endpoint(self, path):
         """
         Constructs and returns a full Dhan API endpoint for a given path.
@@ -56,3 +69,9 @@ def get_chatgpt_endpoint(self):
         Returns the ChatGPT API endpoint.
         """
         return self.chatgpt_endpoint
+
+    def get_weather_endpoint(self):
+        """
+        Returns the Weather API endpoint.
+        """
+        return self.weather_api_endpoint

config/config.py

@@ -19,6 +19,8 @@ def load_config():
             "api_key": os.getenv("DHAN_API_KEY"),
             "secret_key": os.getenv("DHAN_SECRET_KEY"),
             "chatgpt_api_key": os.getenv("CHATGPT_API_KEY"),
+            "weather_api_key": os.getenv("WEATHER_API_KEY"),
+            "weather_api_endpoint": os.getenv("WEATHER_API_ENDPOINT"),
         },
         "strategy": {
             "futures": {

data/data_fetcher.py

@@ -2,6 +2,7 @@
 import logging
 from typing import Dict, Any
 from datetime import datetime
+from data.weather_data_fetcher import WeatherDataFetcher
 
 logger = logging.getLogger(__name__)
 
@@ -21,6 +22,10 @@ def __init__(self, config: Dict[str, Any]):
         self.logger = logging.getLogger(__name__)
         self.logger.info("Initializing DataFetcher")
         self.investment_banking_data = []
+        self.weather_fetcher = WeatherDataFetcher(
+            api_key=config["api"]["weather_api_key"],
+            api_endpoint=config["api"]["weather_api_endpoint"]
+        )
 
     async def process_data_async(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:
         """
@@ -156,3 +161,28 @@ async def process_investment_banking_trades(self, trades_data: Dict[str, Any]) -
         except Exception as e:
             self.logger.error(f"Error processing investment banking trades data: {str(e)}")
             raise
+
+    async def fetch_and_process_weather_data(self, location: str) -> Dict[str, Any]:
+        """
+        Fetch and process weather data for a given location.
+        
+        Args:
+            location: Location for which to fetch weather data
+            
+        Returns:
+            Dictionary containing processed weather data
+        """
+        try:
+            self.logger.info(f"Fetching weather data for location: {location}")
+            weather_data = self.weather_fetcher.fetch_weather_data(location)
+            processed_weather_data = {
+                "location": location,
+                "temperature": weather_data.get("main", {}).get("temp"),
+                "humidity": weather_data.get("main", {}).get("humidity"),
+                "weather": weather_data.get("weather", [{}])[0].get("description")
+            }
+            self.logger.info(f"Weather data processing complete for location: {location}")
+            return processed_weather_data
+        except Exception as e:
+            self.logger.error(f"Error fetching or processing weather data: {str(e)}")
+            return {}

data/weather_data_fetcher.py

@@ -0,0 +1,43 @@
+import requests
+import logging
+
+class WeatherDataFetcher:
+    """
+    Fetches weather data from the Weather API.
+    """
+
+    def __init__(self, api_key: str, api_endpoint: str):
+        """
+        Initialize the WeatherDataFetcher with API key and endpoint.
+        
+        Args:
+            api_key: API key for the Weather API
+            api_endpoint: Endpoint URL for the Weather API
+        """
+        self.api_key = api_key
+        self.api_endpoint = api_endpoint
+        self.logger = logging.getLogger(__name__)
+
+    def fetch_weather_data(self, location: str) -> dict:
+        """
+        Fetch weather data for a given location.
+        
+        Args:
+            location: Location for which to fetch weather data
+            
+        Returns:
+            Dictionary containing weather data
+        """
+        try:
+            self.logger.info(f"Fetching weather data for location: {location}")
+            response = requests.get(
+                self.api_endpoint,
+                params={"q": location, "appid": self.api_key}
+            )
+            response.raise_for_status()
+            weather_data = response.json()
+            self.logger.info(f"Weather data fetched successfully for location: {location}")
+            return weather_data
+        except requests.RequestException as e:
+            self.logger.error(f"Error fetching weather data: {str(e)}")
+            return {}

main.py

@@ -19,6 +19,7 @@
 from data.investment_banking_tracker import InvestmentBankingTracker
 import telegram
 from telegram.ext import Updater, CommandHandler, MessageHandler, Filters
+from data.weather_data_fetcher import WeatherDataFetcher
 
 # Load environment variables and validate required keys
 load_dotenv()
@@ -27,7 +28,9 @@
     "DHAN_SECRET_KEY",
     "CHATGPT_API_KEY",
     "TELEGRAM_BOT_API_KEY",
-    "TELEGRAM_CHAT_ID"
+    "TELEGRAM_CHAT_ID",
+    "WEATHER_API_KEY",
+    "WEATHER_API_ENDPOINT"
 ]
 for var in required_env_vars:
     if not os.getenv(var):
@@ -51,7 +54,8 @@ async def process_market_data(
     chatgpt_integration: ChatGPTIntegration,
     order_executor: OrderExecution,
     notifier: Notifier,
-    investment_banking_tracker: InvestmentBankingTracker
+    investment_banking_tracker: InvestmentBankingTracker,
+    weather_fetcher: WeatherDataFetcher
 ) -> None:
     """
     Process incoming market data asynchronously and execute trading strategies
@@ -65,9 +69,10 @@ async def process_market_data(
         options_task = asyncio.create_task(options_strategy.evaluate_async(processed_data))
         sentiment_task = asyncio.create_task(chatgpt_integration.analyze_sentiment_async(processed_data))
         investment_banking_task = asyncio.create_task(investment_banking_tracker.process_investment_banking_trades(data))
+        weather_task = asyncio.create_task(weather_fetcher.fetch_and_process_weather_data("New York"))
 
-        futures_signal, options_signal, sentiment, _ = await asyncio.gather(
-            futures_task, options_task, sentiment_task, investment_banking_task
+        futures_signal, options_signal, sentiment, _, weather_data = await asyncio.gather(
+            futures_task, options_task, sentiment_task, investment_banking_task, weather_task
         )
 
         # Handle futures signals
@@ -87,6 +92,11 @@ async def process_market_data(
             logger.info(f"AI Sentiment Analysis: {sentiment}")
             await notifier.send_async(f"Market sentiment update: {sentiment}")
 
+        # Process weather data
+        if weather_data:
+            logger.info(f"Weather Data: {weather_data}")
+            await notifier.send_async(f"Weather update: {weather_data}")
+
     except Exception as e:
         logger.error(f"Error processing market data: {str(e)}")
         await notifier.send_async(f"Error in market data processing: {str(e)}")
@@ -119,6 +129,10 @@ async def main():
         notifier = Notifier(config["notifications"], telegram_bot)
         order_executor = OrderExecution(api_key=os.getenv("DHAN_API_KEY"))
         investment_banking_tracker = InvestmentBankingTracker(config["investment_banking"])
+        weather_fetcher = WeatherDataFetcher(
+            api_key=config["api"]["weather_api_key"],
+            api_endpoint=config["api"]["weather_api_endpoint"]
+        )
 
         # Run backtesting with performance metrics
         backtester = Backtester()
@@ -139,7 +153,8 @@ async def on_message(data: Dict) -> None:
                 chatgpt_integration,
                 order_executor,
                 notifier,
-                investment_banking_tracker
+                investment_banking_tracker,
+                weather_fetcher
             )
 
         # Start WebSocket connection

strategies/futures_strategy.py

@@ -4,10 +4,11 @@
 from strategies.base_strategy import BaseStrategy, StrategyResult
 from models.sentiment_analysis import SentimentAnalyzer
 from models.chatgpt_integration import ChatGPTIntegration
+from data.weather_data_fetcher import WeatherDataFetcher
 
 class FuturesStrategy(BaseStrategy):
     """
-    Strategy for trading futures contracts using sentiment analysis and technical indicators.
+    Strategy for trading futures contracts using sentiment analysis, technical indicators, and weather data.
     """
 
     def __init__(self, config: Dict[str, Any]):
@@ -20,10 +21,14 @@ def __init__(self, config: Dict[str, Any]):
         super().__init__(config)
         self.sentiment_analyzer = SentimentAnalyzer(config)
         self.chatgpt = ChatGPTIntegration(config)
+        self.weather_fetcher = WeatherDataFetcher(
+            api_key=config["api"]["weather_api_key"],
+            api_endpoint=config["api"]["weather_api_endpoint"]
+        )
 
     async def generate_signal(self, market_data: Dict[str, Any]) -> StrategyResult:
         """
-        Generate trading signal for futures based on market data and sentiment.
+        Generate trading signal for futures based on market data, sentiment, and weather data.
         
         Args:
             market_data: Dictionary containing processed market data
@@ -44,11 +49,16 @@ async def generate_signal(self, market_data: Dict[str, Any]) -> StrategyResult:
         sentiment = await self.sentiment_analyzer.analyze_sentiment(market_data)
         chatgpt_score = await self.chatgpt.analyze_sentiment_async(market_data)
 
+        # Fetch and process weather data
+        weather_data = await self.weather_fetcher.fetch_weather_data("New York")  # Example location
+        weather_score = self._analyze_weather_data(weather_data)
+
         # Combine signals
         signals = {
             "sentiment": sentiment.score,
             "chatgpt": chatgpt_score,
-            "technical": self._analyze_technical_indicators(market_data)
+            "technical": self._analyze_technical_indicators(market_data),
+            "weather": weather_score
         }
 
         # Calculate final signal and confidence
@@ -100,3 +110,24 @@ def _analyze_technical_indicators(self, data: Dict[str, Any]) -> float:
                 score *= 1.2  # Amplify existing signal
 
         return max(min(score, 1.0), -1.0)  # Clamp between -1 and 1
+
+    def _analyze_weather_data(self, weather_data: Dict[str, Any]) -> float:
+        """
+        Analyze weather data for trading signals.
+        
+        Args:
+            weather_data: Dictionary containing weather data
+            
+        Returns:
+            Weather analysis score between -1 and 1
+        """
+        score = 0.0
+        temperature = weather_data.get("main", {}).get("temp")
+        weather_description = weather_data.get("weather", [{}])[0].get("description")
+
+        if temperature and temperature < 0:  # Example condition for cold weather
+            score -= 0.2
+        if weather_description and "rain" in weather_description.lower():
+            score -= 0.3  # Example condition for rainy weather
+
+        return max(min(score, 1.0), -1.0)  # Clamp between -1 and 1

strategies/options_strategy.py

@@ -2,7 +2,7 @@
 from typing import Dict, Any
 from datetime import datetime
 from .base_strategy import BaseStrategy, StrategyResult
-
+from data.weather_data_fetcher import WeatherDataFetcher
 
 class OptionsStrategy(BaseStrategy):
     """
@@ -20,6 +20,10 @@ def __init__(self, config: Dict[str, Any]):
         super().__init__(config)
         self.logger = logging.getLogger(__name__)
         self.logger.info("Initializing OptionsStrategy")
+        self.weather_fetcher = WeatherDataFetcher(
+            api_key=config["api"]["weather_api_key"],
+            api_endpoint=config["api"]["weather_api_endpoint"]
+        )
 
     async def generate_signal(self, market_data: Dict[str, Any]) -> StrategyResult:
         """
@@ -47,10 +51,15 @@ async def generate_signal(self, market_data: Dict[str, Any]) -> StrategyResult:
         # Analyze technical indicators
         technical_score = self._analyze_technical_indicators(market_data)
 
+        # Fetch and process weather data
+        weather_data = await self.weather_fetcher.fetch_weather_data("New York")  # Example location
+        weather_score = self._analyze_weather_data(weather_data)
+
         # Calculate final signal
         signals = {
             "options": options_score,
-            "technical": technical_score
+            "technical": technical_score,
+            "weather": weather_score
         }
 
         signal = sum(signals.values()) / len(signals)
@@ -140,3 +149,24 @@ def _analyze_technical_indicators(self, data: Dict[str, Any]) -> float:
                 score *= 0.8  # Reduce signal strength in high volatility
 
         return max(min(score, 1.0), -1.0)  # Clamp between -1 and 1
+
+    def _analyze_weather_data(self, weather_data: Dict[str, Any]) -> float:
+        """
+        Analyze weather data for trading signals.
+        
+        Args:
+            weather_data: Dictionary containing weather data
+            
+        Returns:
+            Weather analysis score between -1 and 1
+        """
+        score = 0.0
+        temperature = weather_data.get("main", {}).get("temp")
+        weather_description = weather_data.get("weather", [{}])[0].get("description")
+
+        if temperature and temperature < 0:  # Example condition for cold weather
+            score -= 0.2
+        if weather_description and "rain" in weather_description.lower():
+            score -= 0.3  # Example condition for rainy weather
+
+        return max(min(score, 1.0), -1.0)  # Clamp between -1 and 1