example cleanup

Author: TexasCoding

examples/09_get_check_available_instruments.py

@@ -26,7 +26,7 @@
 from project_x_py.models import Instrument
 
 
-def display_instrument(instrument: Instrument, prefix: str = ""):
+def display_instrument(instrument: Instrument, prefix: str = "") -> None:
     """Display instrument details in a formatted way"""
     print(f"{prefix}┌─ Contract Details ─────────────────────────────")
     print(f"{prefix}│ ID:           {instrument.id}")
@@ -39,7 +39,7 @@ def display_instrument(instrument: Instrument, prefix: str = ""):
     print(f"{prefix}└" + "─" * 47)
 
 
-async def search_and_display(client: ProjectX, symbol: str):
+async def search_and_display(client: ProjectX, symbol: str) -> None:
     """Search for instruments and display results asynchronously"""
     print(f"\n{'=' * 60}")
     print(f"Searching for: '{symbol}'")
@@ -107,13 +107,13 @@ def show_common_symbols():
     print("└─────────┴──────────────────────────────────────────┘")
 
 
-async def get_user_input(prompt):
+async def get_user_input(prompt: str) -> str:
     """Get user input"""
     loop = asyncio.get_event_loop()
     return await loop.run_in_executor(None, input, prompt)
 
 
-async def run_interactive_search(client):
+async def run_interactive_search(client: ProjectX) -> None:
     """Run the interactive search loop."""
     show_common_symbols()
 
@@ -144,7 +144,7 @@ async def run_interactive_search(client):
         await search_and_display(client, symbol.upper())
 
 
-async def main():
+async def main() -> None:
     """Main entry point."""
     print("╔═══════════════════════════════════════════════════════╗")
     print("║   Async ProjectX Instrument Search Interactive Demo   ║")
@@ -161,15 +161,16 @@ async def main():
             print(f"✓ Using account: {client.account_info.name}")
 
             # Show client performance stats periodically
-            async def show_stats():
+            async def show_stats() -> None:
                 while True:
                     await asyncio.sleep(60)  # Every minute
                     stats = await client.get_health_status()
-                    if stats["client_stats"]["api_calls"] > 0:
+                    # Using defined PerformanceStatsResponse keys
+                    if stats["api_calls"] > 0:
                         print(
-                            f"\n[Stats] API calls: {stats['client_stats']['api_calls']}, "
-                            f"Cache hits: {stats['client_stats']['cache_hits']} "
-                            f"({stats['client_stats']['cache_hit_rate']:.1%} hit rate)"
+                            f"\n[Stats] API calls: {stats['api_calls']}, "
+                            f"Cache hits: {stats['cache_hits']} "
+                            f"({(stats['cache_hits'] / max(stats['api_calls'], 1)):.1%} hit rate)"
                         )
 
             # Run stats display in background

examples/11_simplified_data_access.py

@@ -24,7 +24,7 @@
 from project_x_py import TradingSuite
 
 
-async def demonstrate_simplified_access():
+async def demonstrate_simplified_access() -> None:
     """Show the new simplified data access methods."""
 
     # Create trading suite with 3 timeframes
@@ -42,7 +42,7 @@ async def demonstrate_simplified_access():
 
         # 2. Get latest price - much cleaner than get_current_price()
         price = await suite.data.get_latest_price()
-        if price:
+        if price is not None:
             print(f"\n📊 Current Price: ${price:,.2f}")
 
         # 3. Get OHLC as a simple dictionary
@@ -110,7 +110,7 @@ async def demonstrate_simplified_access():
                 print(f"   {tf}: ${close:,.2f} (vol: {volume:,.0f})")
 
 
-async def demonstrate_trading_usage():
+async def demonstrate_trading_usage() -> None:
     """Show how simplified access improves trading logic."""
 
     async with await TradingSuite.create("MNQ") as suite:
@@ -126,7 +126,7 @@ async def demonstrate_trading_usage():
         range_stats = await suite.data.get_price_range(bars=20)
         vol_stats = await suite.data.get_volume_stats(bars=20)
 
-        if price and range_stats and vol_stats:
+        if price is not None and range_stats and vol_stats:
             # Example strategy logic
             print(f"Current Price: ${price:,.2f}")
             print(f"20-bar Range: ${range_stats['range']:,.2f}")
@@ -152,7 +152,7 @@ async def demonstrate_trading_usage():
                 print(f"Price is {range_position:.1%} within the 20-bar range")
 
 
-async def main():
+async def main() -> None:
     """Run all demonstrations."""
     try:
         # Show simplified data access

examples/12_simplified_multi_timeframe.py

@@ -23,10 +23,10 @@ class SimplifiedMTFStrategy:
     def __init__(self, suite: TradingSuite):
         self.suite = suite
         self.data = suite.data  # Direct access to data manager
-        self.position_size = 0
-        self.last_signal_time = None
+        self.position_size: int = 0
+        self.last_signal_time: float | None = None
 
-    async def analyze_market(self):
+    async def analyze_market(self) -> dict | None:
         """Analyze market across multiple timeframes."""
         # Much cleaner than checking data is None and len(data) < X
 
@@ -88,7 +88,7 @@ async def analyze_market(self):
             "volume_strength": volume_stats["relative"] if volume_stats else 1.0,
         }
 
-    async def check_entry_conditions(self):
+    async def check_entry_conditions(self) -> dict | None:
         """Check if all conditions align for entry."""
         analysis = await self.analyze_market()
         if not analysis:
@@ -133,7 +133,7 @@ async def check_entry_conditions(self):
         return None
 
 
-async def run_simplified_mtf_strategy():
+async def run_simplified_mtf_strategy() -> None:
     """Run the simplified multi-timeframe strategy."""
 
     # Create suite with multiple timeframes
@@ -163,12 +163,14 @@ async def run_simplified_mtf_strategy():
                 ohlc = await suite.data.get_ohlc("15min")
 
                 print(f"\n🎯 SIGNAL: {signal['action']}")
-                print(f"   Price: ${price:,.2f}")
+                if price is not None:
+                    print(f"   Price: ${price:,.2f}")
                 print(f"   Confidence: {signal['confidence']:.1%}")
-                print(
-                    f"   15min Bar: O:{ohlc['open']:,.2f} H:{ohlc['high']:,.2f} "
-                    f"L:{ohlc['low']:,.2f} C:{ohlc['close']:,.2f}"
-                )
+                if ohlc is not None:
+                    print(
+                        f"   15min Bar: O:{ohlc['open']:,.2f} H:{ohlc['high']:,.2f} "
+                        f"L:{ohlc['low']:,.2f} C:{ohlc['close']:,.2f}"
+                    )
 
                 # Show multi-timeframe alignment
                 print("\n   Timeframe Alignment:")
@@ -205,7 +207,7 @@ async def run_simplified_mtf_strategy():
         print("\n\n✅ Strategy demonstration complete!")
 
 
-async def compare_verbose_vs_simplified():
+async def compare_verbose_vs_simplified() -> None:
     """Show the difference between verbose and simplified patterns."""
 
     async with await TradingSuite.create("MNQ") as suite:
@@ -259,11 +261,11 @@ async def compare_verbose_vs_simplified():
         # New simplified way
         start = time.time()
         price = await suite.data.get_latest_price()
-        if price:
+        if price is not None:
             print(f"Simplified method: ${price:,.2f} (took {time.time() - start:.3f}s)")
 
 
-async def main():
+async def main() -> None:
     """Run all demonstrations."""
     try:
         # Run simplified multi-timeframe strategy

examples/14_phase4_comprehensive_test.py

@@ -31,7 +31,9 @@ def __init__(self, suite: TradingSuite):
         self.profit_target_ticks = 20
         self.stop_loss_ticks = 10
 
-    async def analyze_market(self) -> dict | None:
+    async def analyze_market(
+        self,
+    ) -> dict[str, float | int | str | dict[str, float]] | None:
         """Analyze market using simplified data access."""
         # Use new get_data_or_none for cleaner code
         data = await self.data.get_data_or_none("5min", min_bars=50)
@@ -47,7 +49,12 @@ async def analyze_market(self) -> dict | None:
         price_range = await self.data.get_price_range(bars=20)
         volume_stats = await self.data.get_volume_stats(bars=20)
 
-        if not all([current_price, ohlc, price_range, volume_stats]):
+        if (
+            current_price is None
+            or ohlc is None
+            or price_range is None
+            or volume_stats is None
+        ):
             return None
 
         # Analyze trend
@@ -56,21 +63,30 @@ async def analyze_market(self) -> dict | None:
         atr = float(data["atr_14"][-1])
 
         # Price position within range
-        price_position = (current_price - price_range["low"]) / price_range["range"]
+        price_position = (float(current_price) - float(price_range["low"])) / float(
+            price_range["range"]
+        )
 
         return {
-            "price": current_price,
-            "trend": "bullish" if current_price > sma20 else "bearish",
-            "trend_strength": abs(current_price - sma20) / sma20,
+            "price": float(current_price),
+            "trend": "bullish" if float(current_price) > sma20 else "bearish",
+            "trend_strength": abs(float(current_price) - sma20) / sma20,
             "rsi": rsi,
             "atr": atr,
             "price_position": price_position,
-            "volume_relative": volume_stats["relative"],
-            "range": price_range["range"],
-            "ohlc": ohlc,
+            "volume_relative": float(volume_stats["relative"]),
+            "range": float(price_range["range"]),
+            "ohlc": {
+                "open": float(ohlc["open"]),
+                "high": float(ohlc["high"]),
+                "low": float(ohlc["low"]),
+                "close": float(ohlc["close"]),
+            },
         }
 
-    async def check_positions(self) -> dict:
+    async def check_positions(
+        self,
+    ) -> dict[str, float | int | list[dict[str, float | int | str]]]:
         """Check positions using enhanced model properties."""
         positions = await self.positions.get_all_positions()
 
@@ -83,7 +99,7 @@ async def check_positions(self) -> dict:
         }
 
         current_price = await self.data.get_latest_price()
-        if not current_price:
+        if current_price is None:
             return position_summary
 
         for pos in positions:
@@ -96,7 +112,7 @@ async def check_positions(self) -> dict:
             position_summary["total_exposure"] += pos.total_cost
 
             # Calculate P&L using the unrealized_pnl method
-            pnl = pos.unrealized_pnl(current_price, tick_value=5.0)  # MNQ tick value
+            pnl = pos.unrealized_pnl(float(current_price), tick_value=5.0)
 
             position_summary["positions"].append(
                 {
@@ -113,7 +129,9 @@ async def check_positions(self) -> dict:
 
         return position_summary
 
-    async def check_orders(self) -> dict:
+    async def check_orders(
+        self,
+    ) -> dict[str, float | int | list[dict[str, float | int | str]]]:
         """Check orders using enhanced model properties."""
         orders = await self.orders.search_open_orders()
 
@@ -145,13 +163,17 @@ async def check_orders(self) -> dict:
                     "size": order.size,
                     "remaining": order.remaining_size,  # New property
                     "filled_pct": order.filled_percent,  # New property
-                    "price": order.limitPrice or order.stopPrice,
+                    "price": float(order.limitPrice)
+                    if order.limitPrice
+                    else float(order.stopPrice)
+                    if order.stopPrice
+                    else 0.0,
                 }
             )
 
         return order_summary
 
-    async def execute_strategy(self):
+    async def execute_strategy(self) -> None:
         """Execute trading strategy using all Phase 4 improvements."""
         print("\n=== Strategy Execution ===")
 
@@ -221,7 +243,9 @@ async def execute_strategy(self):
             )
             print(f"  Reason: {signal['reason']}")
 
-    def _generate_signal(self, analysis: dict, positions: dict) -> dict | None:
+    def _generate_signal(
+        self, analysis: dict, positions: dict
+    ) -> dict[str, float | str] | None:
         """Generate trading signal based on analysis."""
         # No signal if we have max positions
         if positions["total_positions"] >= self.max_position_size:
@@ -264,7 +288,7 @@ def _generate_signal(self, analysis: dict, positions: dict) -> dict | None:
         return None
 
 
-async def demonstrate_phase4_improvements():
+async def demonstrate_phase4_improvements() -> None:
     """Demonstrate all Phase 4 improvements in action."""
 
     async with await TradingSuite.create(
@@ -365,7 +389,7 @@ async def demonstrate_phase4_improvements():
         print("\n✅ Overall: Cleaner, more readable, less error-prone code!")
 
 
-async def main():
+async def main() -> None:
     """Run Phase 4 comprehensive test."""
     try:
         await demonstrate_phase4_improvements()