Add support for arbitrary filter functions

TwitchChannelPointsMiner/classes/Twitch.py

@@ -396,7 +396,7 @@ def load_channel_points_context(self, streamer):
                 self.claim_bonus(streamer, community_points["availableClaim"]["id"])
 
     def make_predictions(self, event):
-        decision = event.bet.calculate(event.streamer.channel_points)
+        decision = event.bet.calculate(event.streamer.channel_points, event.title)
         selector_index = 0 if decision["choice"] == "A" else 1
 
         logger.info(
@@ -407,7 +407,7 @@ def make_predictions(self, event):
             },
         )
         if event.status == "ACTIVE":
-            skip, compared_value = event.bet.skip()
+            skip, compared_value = event.bet.skip(event.title)
             if skip is True:
                 logger.info(
                     f"Skip betting for the event {event}",

TwitchChannelPointsMiner/classes/entities/Bet.py

@@ -1,6 +1,7 @@
 import copy
 from enum import Enum, auto
 from random import uniform
+from typing import Callable
 
 from millify import millify
 
@@ -12,6 +13,7 @@ class Strategy(Enum):
     HIGH_ODDS = auto()
     PERCENTAGE = auto()
     SMART = auto()
+    DYNAMIC = auto()
 
     def __str__(self):
         return self.name
@@ -75,6 +77,7 @@ class BetSettings(object):
         "minimum_points",
         "stealth_mode",
         "filter_condition",
+        "filter_func",
         "delay",
         "delay_mode",
     ]
@@ -88,6 +91,7 @@ def __init__(
         minimum_points: int = None,
         stealth_mode: bool = None,
         filter_condition: FilterCondition = None,
+        filter_func: Callable[[list, str], str] = None,
         delay: float = None,
         delay_mode: DelayMode = None,
     ):
@@ -98,6 +102,7 @@ def __init__(
         self.minimum_points = minimum_points
         self.stealth_mode = stealth_mode
         self.filter_condition = filter_condition
+        self.filter_func = filter_func
         self.delay = delay
         self.delay_mode = delay_mode
 
@@ -215,8 +220,10 @@ def __clear_outcomes(self):
     def __return_choice(self, key) -> str:
         return "A" if self.outcomes[0][key] > self.outcomes[1][key] else "B"
 
-    def skip(self) -> bool:
-        if self.settings.filter_condition is not None:
+    def skip(self, title) -> bool:
+        if self.settings.filter_func is not None:
+            return self.settings.filter_func(self.outcomes, title) == None, 0
+        elif self.settings.filter_condition is not None:
             # key == by , condition == where
             key = self.settings.filter_condition.by
             condition = self.settings.filter_condition.where
@@ -252,7 +259,7 @@ def skip(self) -> bool:
         else:
             return False, 0  # Default don't skip the bet
 
-    def calculate(self, balance: int) -> dict:
+    def calculate(self, balance: int, title: str) -> dict:
         self.decision = {"choice": None, "amount": 0, "id": None}
         if self.settings.strategy == Strategy.MOST_VOTED:
             self.decision["choice"] = self.__return_choice(OutcomeKeys.TOTAL_USERS)
@@ -270,6 +277,8 @@ def calculate(self, balance: int) -> dict:
                 if difference < self.settings.percentage_gap
                 else self.__return_choice(OutcomeKeys.TOTAL_USERS)
             )
+        elif self.settings.strategy == Strategy.DYNAMIC:
+            self.decision["choice"] = self.settings.filter_func(self.outcomes, title)
 
         if self.decision["choice"] is not None:
             index = char_decision_as_index(self.decision["choice"])