Skip to content

Commit

Permalink
Apply filter on betting, skip some events if the criteria doesn't mee…
Browse files Browse the repository at this point in the history 
…t - #29
  • Loading branch information
@Tkd-Alex
Tkd-Alex committed Feb 1, 2021
1 parent 64623ca commit c737132
Show file tree
Hide file tree
Showing 2 changed files with 144 additions and 67 deletions.
64 changes: 36 additions & 28 deletions TwitchChannelPointsMiner/classes/TwitchBrowser.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -298,41 +298,49 @@ def __bet_chains_methods(self, event) -> bool:
return False

def place_bet(self, event: EventPrediction):
logger.info(
f"Place bet for {event} owned by {event.streamer}",
extra={"emoji": ":wrench:"},
)
if event.status == "ACTIVE":
self.currently_is_betting = self.start_bet(event)
if event.box_fillable and self.currently_is_betting:
decision = event.bet.calculate(event.streamer.channel_points)
if decision["choice"] is not None:
selector_index = 1 if decision["choice"] == "A" else 2
logger.info(
f"Decision: {event.bet.get_outcome(selector_index - 1)}",
extra={"emoji": ":wrench:"},
)

try:
decision = event.bet.calculate(event.streamer.channel_points)
if event.bet.skip() is True:
logger.info(f"Skip betting for the event {event}")
logger.info(f"Skip settings {event.bet.settings.filter_condition}")
else:
self.currently_is_betting = self.start_bet(event)
if event.box_fillable and self.currently_is_betting:
if decision["choice"] is not None:
selector_index = 1 if decision["choice"] == "A" else 2
logger.info(
f"Going to write: {_millify(decision['amount'])} channel points on input {decision['choice']}",
f"Decision: {event.bet.get_outcome(selector_index - 1)}",
extra={"emoji": ":wrench:"},
)
if (
self.__send_text_on_bet(
event, selector_index, decision["amount"]
)
is True
):

try:
logger.info(
f"Going to place the bet for {event}",
f"Going to write: {_millify(decision['amount'])} channel points on input {decision['choice']}",
extra={"emoji": ":wrench:"},
)
if self.__click_on_vote(event, selector_index) is True:
event.bet_placed = True
time.sleep(random.uniform(5, 10))
except Exception:
logger.error("Exception raised", exc_info=True)
else:
logger.info(
f"Sorry, unable to complete the bet. Event box fillable: {event.box_fillable}, the browser is betting: {self.currently_is_betting}"
)
if (
self.__send_text_on_bet(
event, selector_index, decision["amount"]
)
is True
):
logger.info(
f"Going to place the bet for {event}",
extra={"emoji": ":wrench:"},
)
if self.__click_on_vote(event, selector_index) is True:
event.bet_placed = True
time.sleep(random.uniform(5, 10))
except Exception:
logger.error("Exception raised", exc_info=True)
else:
logger.info(
f"Sorry, unable to complete the bet. Event box fillable: {event.box_fillable}, the browser is betting: {self.currently_is_betting}"
)
else:
logger.info(
f"Oh no! The event is not active anymore! Current status: {event.status}",
Expand Down
147 changes: 108 additions & 39 deletions TwitchChannelPointsMiner/classes/entities/Bet.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,33 @@ class Strategy(Enum):
SMART = auto()


class Condition(Enum):
GT = auto()
LT = auto()
GTE = auto()
LTE = auto()


class OutcomeKeys(object):
PERCENTAGE_USERS = "percentage_users"
ODDS_PERCENTAGE = "odds_percentage"
ODDS = "odds"
TOP_POINTS = "top_points"
TOTAL_USERS = "total_users"
TOTAL_POINTS = "total_points"


class FilterCondition(object):
def __init__(self, key=None, condition=None, value=None, decision=None):
self.key = key
self.condition = condition
self.value = value
self.decision = decision

def __repr__(self):
return f"FilterCondition(Key={self.key}, Condition={self.condition}, Value={self.value}, Index={self.index})"


class BetSettings:
def __init__(
self,
Expand All @@ -25,11 +52,13 @@ def __init__(
percentage_gap: int = None,
max_points: int = None,
stealth_mode: bool = None,
filter_condition: FilterCondition = None,
):
self.strategy = strategy
self.percentage = percentage
self.percentage_gap = percentage_gap
self.max_points = max_points
self.filter_condition = filter_condition

def default(self):
self.strategy = self.strategy if not None else Strategy.SMART
Expand All @@ -52,41 +81,47 @@ def __init__(self, outcomes: list, settings: BetSettings):
self.settings = settings

def update_outcomes(self, outcomes):
self.outcomes[0]["total_users"] = int(outcomes[0]["total_users"])
self.outcomes[1]["total_users"] = int(outcomes[1]["total_users"])
self.outcomes[0]["total_points"] = int(outcomes[0]["total_points"])
self.outcomes[1]["total_points"] = int(outcomes[1]["total_points"])
for index in range(0, len(self.outcomes)):
self.outcomes[index][OutcomeKeys.TOTAL_USERS] = int(
outcomes[index][OutcomeKeys.TOTAL_USERS]
)
self.outcomes[index][OutcomeKeys.TOTAL_POINTS] = int(
outcomes[index][OutcomeKeys.TOTAL_POINTS]
)

outcomes[0]["top_predictors"] = sorted(
outcomes[0]["top_predictors"], key=lambda x: x["points"], reverse=True
)
outcomes[1]["top_predictors"] = sorted(
outcomes[1]["top_predictors"], key=lambda x: x["points"], reverse=True
)
self.outcomes[0]["top_points"] = outcomes[0]["top_predictors"]["points"]
self.outcomes[1]["top_points"] = outcomes[1]["top_predictors"]["points"]
outcomes[index]["top_predictors"] = sorted(
outcomes[index]["top_predictors"],
key=lambda x: x["points"],
reverse=True,
)

top_points = outcomes[index]["top_predictors"]["points"]
self.outcomes[index][OutcomeKeys.TOP_POINTS] = top_points

self.total_users = (
self.outcomes[0]["total_users"] + self.outcomes[1]["total_users"]
self.outcomes[0][OutcomeKeys.TOTAL_USERS]
+ self.outcomes[1][OutcomeKeys.TOTAL_USERS]
)
self.total_points = (
self.outcomes[0]["total_points"] + self.outcomes[1]["total_points"]
self.outcomes[0][OutcomeKeys.TOTAL_POINTS]
+ self.outcomes[1][OutcomeKeys.TOTAL_POINTS]
)

if (
self.total_users > 0
and self.outcomes[0]["total_points"] > 0
and self.outcomes[1]["total_points"] > 0
and self.outcomes[0][OutcomeKeys.TOTAL_POINTS] > 0
and self.outcomes[1][OutcomeKeys.TOTAL_POINTS] > 0
):
for index in range(0, len(self.outcomes)):
self.outcomes[index]["percentage_users"] = float_round(
(100 * self.outcomes[index]["total_users"]) / self.total_users
self.outcomes[index][OutcomeKeys.PERCENTAGE_USERS] = float_round(
(100 * self.outcomes[index][OutcomeKeys.TOTAL_USERS])
/ self.total_users
)
self.outcomes[index]["odds"] = float_round(
self.total_points / self.outcomes[index]["total_points"]
self.outcomes[index][OutcomeKeys.ODDS] = float_round(
self.total_points / self.outcomes[index][OutcomeKeys.TOTAL_POINTS]
)
self.outcomes[index]["odds_percentage"] = float_round(
100 / self.outcomes[index]["odds"]
self.outcomes[index][OutcomeKeys.ODDS_PERCENTAGE] = float_round(
100 / self.outcomes[index][OutcomeKeys.ODDS]
)

self.__clear_outcomes()
Expand All @@ -96,48 +131,81 @@ def __repr__(self):

def get_outcome(self, index):
outcome = self.outcomes[index]
return f"{outcome['title']} ({outcome['color']}), Points: {millify(outcome['total_points'])}, Users: {millify(outcome['total_users'])} ({outcome['percentage_users']}%), Odds: {outcome['odds']} ({outcome['odds_percentage']}%)"
return f"{outcome['title']} ({outcome['color']}), Points: {millify(outcome[OutcomeKeys.TOTAL_POINTS])}, Users: {millify(outcome[OutcomeKeys.TOTAL_USERS])} ({outcome[OutcomeKeys.PERCENTAGE_USERS]}%), Odds: {outcome[OutcomeKeys.ODDS]} ({outcome[OutcomeKeys.ODDS_PERCENTAGE]}%)"

def __clear_outcomes(self):
for index in range(0, len(self.outcomes)):
keys = copy.deepcopy(list(self.outcomes[index].keys()))
for key in keys:
if key not in [
"total_users",
"total_points",
"top_points",
"percentage_users",
"odds",
"odds_percentage",
OutcomeKeys.TOTAL_USERS,
OutcomeKeys.TOTAL_POINTS,
OutcomeKeys.TOP_POINTS,
OutcomeKeys.PERCENTAGE_USERS,
OutcomeKeys.ODDS,
OutcomeKeys.ODDS_PERCENTAGE,
"title",
"color",
"id",
]:
del self.outcomes[index][key]
for key in ["percentage_users", "odds", "odds_percentage", "top_points"]:
for key in [
OutcomeKeys.PERCENTAGE_USERS,
OutcomeKeys.ODDS,
OutcomeKeys.ODDS_PERCENTAGE,
OutcomeKeys.TOP_POINTS,
]:
if key not in self.outcomes[index]:
self.outcomes[index][key] = 0

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:
key = self.settings.filter_condition.key
condition = self.settings.filter_condition.condition
value = self.settings.filter_condition.value

compared_value = (
(self.outcomes[0][key] + self.outcomes[1][key])
if self.settings.filter_condition.decision is False
else self.outcomes[char_decision_as_index(self.decision["choice"])][key]
)
logger.info(
f"Filter applied on this bet: {compared_value} {condition} {value}"
)
if condition == Condition.GT:
if compared_value > value:
return True
elif condition == Condition.LT:
if compared_value < value:
return True
elif condition == Condition.GTE:
if compared_value >= value:
return True
elif condition == Condition.LTE:
if compared_value <= value:
return True
return False

def calculate(self, balance: int) -> dict:
self.decision = {"choice": None, "amount": 0, "id": None}
if self.settings.strategy == Strategy.MOST_VOTED:
self.decision["choice"] = self.__return_choice("total_users")
self.decision["choice"] = self.__return_choice(OutcomeKeys.TOTAL_USERS)
elif self.settings.strategy == Strategy.HIGH_ODDS:
self.decision["choice"] = self.__return_choice("odds")
self.decision["choice"] = self.__return_choice(OutcomeKeys.ODDS)
elif self.settings.strategy == Strategy.PERCENTAGE:
self.decision["choice"] = self.__return_choice("odds_percentage")
self.decision["choice"] = self.__return_choice(OutcomeKeys.ODDS_PERCENTAGE)
elif self.settings.strategy == Strategy.SMART:
difference = abs(
self.outcomes[0]["percentage_users"]
- self.outcomes[1]["percentage_users"]
self.outcomes[0][OutcomeKeys.PERCENTAGE_USERS]
- self.outcomes[1][OutcomeKeys.PERCENTAGE_USERS]
)
self.decision["choice"] = (
self.__return_choice("odds")
self.__return_choice(OutcomeKeys.ODDS)
if difference < self.settings.percentage_gap
else self.__return_choice("total_users")
else self.__return_choice(OutcomeKeys.TOTAL_USERS)
)

if self.decision["choice"] is not None:
Expand All @@ -149,10 +217,11 @@ def calculate(self, balance: int) -> dict:
)
if (
self.settings.stealth_mode is True
and self.decision["amount"] >= self.outcomes[index]["top_points"]
and self.decision["amount"]
>= self.outcomes[index][OutcomeKeys.TOP_POINTS]
):
reduce_amount = uniform(1, 5)
self.decision["amount"] = (
self.outcomes[index]["top_points"] - reduce_amount
self.outcomes[index][OutcomeKeys.TOP_POINTS] - reduce_amount
)
return self.decision

0 comments on commit c737132

Please sign in to comment.