diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 9aba8dd..33ffb73 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -18,7 +18,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ["3.12"] + python-version: ["3.13"] steps: - uses: actions/checkout@v4 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b56586d..c75d60e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -10,9 +10,9 @@ repos: --fix, --unsafe-fixes, --exit-non-zero-on-fix, - --target-version, "py312", + --target-version, "py313", --select, "ALL", - --ignore, "ANN101,C90,D,E501,ERA001,FBT,PLR091,PLR2004,RUF001,S101,T20" + --ignore, "A004,C901,CPY001,D,DOC,E501,ERA001,INP001,PLR091,PLR2004,RUF001,S101,T201" ] - id: ruff-format diff --git a/market_arbitrage.py b/market_arbitrage.py index 985f6e4..29d5cd4 100644 --- a/market_arbitrage.py +++ b/market_arbitrage.py @@ -1,6 +1,8 @@ # Objective: find market arbitrages, e.g. sell a pack for more (fee excluded) than the cost to craft it (fee included). +from typing import Annotated + from src.inventory_utils import create_then_sell_booster_packs_for_batch from src.market_arbitrage_utils import ( convert_arbitrages_for_batch_create_then_sell, @@ -15,12 +17,13 @@ def apply_workflow( + *, # enforce keyword arguments retrieve_listings_from_scratch: bool = True, retrieve_market_orders_online: bool = True, enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, automatically_create_then_sell_booster_packs: bool = False, - profit_threshold: float = 0.01, # profit in euros + profit_threshold: Annotated[float, "profit in euros"] = 0.01, quick_check_with_tracked_booster_packs: bool = False, enforce_update_of_marketability_status: bool = False, from_javascript: bool = False, @@ -34,7 +37,9 @@ def apply_workflow( retrieve_market_orders_online = True print( - f"Overwriting two arguments:\n\ti) retrieve listings: {retrieve_listings_from_scratch},\n\tii) retrieve market orders: {retrieve_market_orders_online}.", + f"Overwriting two arguments:\n" + f"\ti) retrieve listings: {retrieve_listings_from_scratch},\n" + f"\tii) retrieve market orders: {retrieve_market_orders_online}.", ) filtered_badge_data = get_filtered_badge_data( @@ -107,7 +112,7 @@ def apply_workflow( profit_threshold=profit_threshold, ) - creation_results, sale_results = create_then_sell_booster_packs_for_batch( + _creation_results, _sale_results = create_then_sell_booster_packs_for_batch( price_dict_for_listing_hashes, focus_on_marketable_items=True, profile_id=profile_id, diff --git a/market_arbitrage_with_foil_cards.py b/market_arbitrage_with_foil_cards.py index 60fdf37..e5754fc 100644 --- a/market_arbitrage_with_foil_cards.py +++ b/market_arbitrage_with_foil_cards.py @@ -33,6 +33,7 @@ def apply_workflow_for_foil_cards( + *, retrieve_listings_from_scratch: bool = False, price_threshold_in_cents_for_a_foil_card: float | None = None, retrieve_gem_price_from_scratch: bool = False, diff --git a/market_buzz_detector.py b/market_buzz_detector.py index 56a21b4..08a9279 100644 --- a/market_buzz_detector.py +++ b/market_buzz_detector.py @@ -1,5 +1,7 @@ # Objective: detect the buzz, for games which I do not own yet, i.e. find packs which are likely to have high bid orders +from typing import Final + from src.market_arbitrage_utils import ( filter_out_badges_with_low_sell_price, find_badge_arbitrages, @@ -17,17 +19,22 @@ from src.market_search import load_all_listings, update_all_listings from src.market_utils import filter_out_dubious_listing_hashes +DEFAULT_MIN_SELL_PRICE: Final[int] = 30 +DEFAULT_MIN_NUM_LISTINGS: Final[int] = 3 +DEFAULT_NUM_PACKS_TO_DISPLAY: Final[int] = 10 + def main( + *, retrieve_listings_from_scratch: bool = False, retrieve_market_orders_online: bool = False, force_update_from_steam_card_exchange: bool = False, enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, use_a_constant_price_threshold: bool = False, - min_sell_price: float = 30, - min_num_listings: int = 3, - num_packs_to_display: int = 10, + min_sell_price: int = DEFAULT_MIN_SELL_PRICE, + min_num_listings: int = DEFAULT_MIN_NUM_LISTINGS, + num_packs_to_display: int = DEFAULT_NUM_PACKS_TO_DISPLAY, verbose: bool = False, ) -> None: # Load list of all listing hashes @@ -94,7 +101,7 @@ def main( ( marketable_market_order_dict, - unknown_market_order_dict, + _unknown_market_order_dict, ) = filter_out_unmarketable_packs(market_order_dict) # Sort by bid value diff --git a/market_gamble_detector.py b/market_gamble_detector.py index 12969d3..e87c57d 100644 --- a/market_gamble_detector.py +++ b/market_gamble_detector.py @@ -18,6 +18,7 @@ sort_according_to_buzz, ) from src.market_gamble_utils import ( + DropRateEstimates, count_listing_hashes_per_app_id, enumerate_item_rarity_patterns, filter_out_candidates_whose_ask_price_is_below_threshold, @@ -39,14 +40,14 @@ def main( + *, look_for_profile_backgrounds: bool = True, # if True, profile backgrounds, otherwise, emoticons. retrieve_listings_from_scratch: bool = False, retrieve_listings_with_another_rarity_tag_from_scratch: bool = False, retrieve_market_orders_online: bool = True, focus_on_listing_hashes_never_seen_before: bool = True, price_threshold_in_cents: float | None = None, - drop_rate_estimates_for_common_rarity: dict[tuple[int, int, int], float] - | None = None, + drop_rate_estimates_for_common_rarity: DropRateEstimates | None = None, num_packs_to_display: int = 10, enforce_cooldown: bool = True, allow_to_skip_dummy_data: bool = False, @@ -134,10 +135,10 @@ def main( market_order_dict = get_market_orders( filtered_badge_data, - retrieve_market_orders_online, - focus_on_listing_hashes_never_seen_before, - listing_details_output_file_name, - market_order_output_file_name, + retrieve_market_orders_online=retrieve_market_orders_online, + focus_on_listing_hashes_never_seen_before=focus_on_listing_hashes_never_seen_before, + listing_details_output_file_name=listing_details_output_file_name, + market_order_output_file_name=market_order_output_file_name, enforce_cooldown=enforce_cooldown, allow_to_skip_dummy_data=allow_to_skip_dummy_data, verbose=verbose, @@ -147,7 +148,7 @@ def main( ( marketable_market_order_dict, - unknown_market_order_dict, + _unknown_market_order_dict, ) = filter_out_unmarketable_packs(market_order_dict) # Sort by bid value diff --git a/src/api_utils.py b/src/api_utils.py new file mode 100644 index 0000000..284055b --- /dev/null +++ b/src/api_utils.py @@ -0,0 +1,32 @@ +from typing import Final + +from src.utils import get_cushioned_cooldown_in_seconds + +INTER_REQUEST_COOLDOWN_FIELD: Final[str] = "cooldown_between_each_request" + + +def get_rate_limits( + api_type: str, + *, + has_secured_cookie: bool = False, +) -> dict[str, int]: + if has_secured_cookie: + base_limits = { + "market_order": {"queries": 50, "minutes": 1}, + "market_search": {"queries": 50, "minutes": 1}, + "market_listing": {"queries": 25, "minutes": 3}, + } + else: + base_limits = { + "market_order": {"queries": 25, "minutes": 5}, + "market_search": {"queries": 25, "minutes": 5}, + "market_listing": {"queries": 25, "minutes": 5}, + } + + limits = base_limits[api_type] + + return { + "max_num_queries": limits["queries"], + "cooldown": get_cushioned_cooldown_in_seconds(num_minutes=limits["minutes"]), + INTER_REQUEST_COOLDOWN_FIELD: 0, + } diff --git a/src/batch_create_packs.py b/src/batch_create_packs.py index ece2333..ed5b28c 100644 --- a/src/batch_create_packs.py +++ b/src/batch_create_packs.py @@ -15,6 +15,7 @@ def get_manually_selected_app_ids() -> list[str]: def filter_app_ids_based_on_badge_data( manually_selected_app_ids: list[str], + *, check_ask_price: bool = False, filtered_badge_data: dict[str, dict] | None = None, ) -> tuple[list[str], dict[str, dict]]: @@ -44,6 +45,7 @@ def filter_app_ids_based_on_badge_data( def create_packs_for_app_ids( manually_selected_app_ids: list[str], filtered_badge_data: dict[str, dict] | None = None, + *, check_ask_price: bool = False, is_a_simulation: bool = True, # Caveat: if False, then packs will be crafted, which costs money! @@ -111,6 +113,7 @@ def create_packs_for_app_ids( def main( + *, retrieve_listings_from_scratch: bool = False, # Set to True & run once if you get "No match found for" games you own. is_a_simulation: bool = True, # Caveat: if False, then packs will be crafted, which costs money! @@ -140,7 +143,7 @@ def main( from_javascript=from_javascript, ) - creation_results, next_creation_times = create_packs_for_app_ids( + _creation_results, _next_creation_times = create_packs_for_app_ids( manually_selected_app_ids, filtered_badge_data=filtered_badge_data, check_ask_price=check_ask_price, diff --git a/src/creation_time_utils.py b/src/creation_time_utils.py index ce663e4..f09fc40 100644 --- a/src/creation_time_utils.py +++ b/src/creation_time_utils.py @@ -20,6 +20,7 @@ def load_next_creation_time_data( def fill_in_badges_with_next_creation_times_loaded_from_disk( aggregated_badge_data: dict[str, dict], + *, verbose: bool = True, ) -> dict[str, dict]: next_creation_times_loaded_from_disk = load_next_creation_time_data() @@ -63,7 +64,7 @@ def to_timestamp(date: datetime.datetime) -> int: return int(date.timestamp()) -def get_creation_time_format(prepend_year: bool = False) -> str: +def get_creation_time_format(*, prepend_year: bool = False) -> str: # Reference: https://docs.python.org/3/library/time.html#time.strftime # The format used in: "14 Sep @ 10:48pm" @@ -98,6 +99,7 @@ def prepend_year_to_time_as_str( def get_time_struct_from_str( formatted_time_as_str: str, + *, use_current_year: bool = False, ) -> datetime.datetime: if use_current_year: diff --git a/src/download_steam_card_exchange.py b/src/download_steam_card_exchange.py index a05c6d3..0afc89a 100644 --- a/src/download_steam_card_exchange.py +++ b/src/download_steam_card_exchange.py @@ -40,6 +40,7 @@ def save_data_from_steam_card_exchange( def download_data_from_steam_card_exchange( steam_card_exchange_file_name: str | None = None, + *, save_to_disk: bool = True, ) -> dict | None: if steam_card_exchange_file_name is None: @@ -97,6 +98,7 @@ def compute_gem_amount_required_to_craft_booster_pack(num_cards_per_set: int) -> def parse_data_from_steam_card_exchange( response: dict | None = None, + *, force_update_from_steam_card_exchange: bool = False, steam_card_exchange_file_name: str | None = None, ) -> dict[str, dict]: @@ -140,7 +142,7 @@ def parse_data_from_steam_card_exchange( return dico -def main(force_update: bool = False) -> bool: +def main(*, force_update: bool = False) -> bool: if force_update: response = download_data_from_steam_card_exchange() else: diff --git a/src/drop_rate_estimates.py b/src/drop_rate_estimates.py index e1b9a3c..910146a 100644 --- a/src/drop_rate_estimates.py +++ b/src/drop_rate_estimates.py @@ -29,7 +29,10 @@ def clamp_proportion(input_proportion: float) -> float: return min(1.0, max(0.0, input_proportion)) -def get_drop_rate_estimates_based_on_item_rarity_pattern(verbose: bool = True) -> dict: +def get_drop_rate_estimates_based_on_item_rarity_pattern( + *, + verbose: bool = True, +) -> dict: # Drop-rate estimates conditionally on the item rarity pattern C/UC/R (the numbers of possible items of each rarity) drop_rate_estimates: dict = {} @@ -162,7 +165,7 @@ def get_drop_rate_estimates_based_on_item_rarity_pattern(verbose: bool = True) - return drop_rate_estimates -def get_drop_rate_estimates(verbose: bool = True) -> dict: +def get_drop_rate_estimates(*, verbose: bool = True) -> dict: # Drop-rate estimates conditionally on the category (profile backgrounds, emoticons) drop_rate_estimates: dict = {} diff --git a/src/free_games_with_trading_cards.py b/src/free_games_with_trading_cards.py index 6c3bb49..04dd3a9 100644 --- a/src/free_games_with_trading_cards.py +++ b/src/free_games_with_trading_cards.py @@ -38,7 +38,7 @@ def download_user_data() -> dict | None: return result -def download_owned_apps(verbose: bool = True) -> list[str]: +def download_owned_apps(*, verbose: bool = True) -> list[str]: result = download_user_data() owned_apps = result["rgOwnedApps"] if result else [] @@ -49,7 +49,7 @@ def download_owned_apps(verbose: bool = True) -> list[str]: return [str(i) for i in owned_apps] -def download_free_apps(method: str = "price", verbose: bool = True) -> set[str]: +def download_free_apps(method: str = "price", *, verbose: bool = True) -> set[str]: if method == "price": data = steamspypi.load() @@ -81,7 +81,7 @@ def download_free_apps(method: str = "price", verbose: bool = True) -> set[str]: return free_apps -def load_apps_with_trading_cards(verbose: bool = True) -> list[str]: +def load_apps_with_trading_cards(*, verbose: bool = True) -> list[str]: all_listings = load_all_listings() apps_with_trading_cards = [ @@ -97,6 +97,7 @@ def load_apps_with_trading_cards(verbose: bool = True) -> list[str]: def load_free_apps_with_trading_cards( free_apps: set[str] | None = None, list_of_methods: list[str] | None = None, + *, verbose: bool = True, ) -> set[str]: if list_of_methods is None: @@ -123,7 +124,7 @@ def load_free_apps_with_trading_cards( return free_apps_with_trading_cards -def load_file(file_name: str, verbose: bool = True) -> list[str]: +def load_file(file_name: str, *, verbose: bool = True) -> list[str]: with Path(file_name).open(encoding="utf-8") as f: data = [line.strip() for line in f] @@ -189,6 +190,7 @@ def write_to_file( file_name: str, asf_username: str, group_size: int = 25, + *, verbose: bool = True, ) -> None: output = group_concatenate_to_str( diff --git a/src/inventory_utils.py b/src/inventory_utils.py index b43c7e5..9c0b66f 100644 --- a/src/inventory_utils.py +++ b/src/inventory_utils.py @@ -64,21 +64,20 @@ def load_steam_inventory_from_disk(profile_id: str | None = None) -> dict | None def load_steam_inventory( profile_id: str | None = None, + *, update_steam_inventory: bool = False, ) -> dict | None: if profile_id is None: profile_id = get_my_steam_profile_id() if update_steam_inventory: - steam_inventory = download_steam_inventory(profile_id, save_to_disk=True) - else: - steam_inventory = load_steam_inventory_from_disk(profile_id=profile_id) - - return steam_inventory + return download_steam_inventory(profile_id, save_to_disk=True) + return load_steam_inventory_from_disk(profile_id) def download_steam_inventory( profile_id: str | None = None, + *, save_to_disk: bool = True, ) -> dict | None: if profile_id is None: @@ -131,6 +130,7 @@ def get_steam_booster_pack_creation_url() -> str: def get_booster_pack_creation_parameters( app_id: str, session_id: str, + *, is_marketable: bool = True, ) -> dict[str, str]: booster_pack_creation_parameters = {} @@ -147,6 +147,7 @@ def get_booster_pack_creation_parameters( def create_booster_pack( app_id: str, + *, is_marketable: bool = True, verbose: bool = True, ) -> dict | None: @@ -209,17 +210,15 @@ def get_market_sell_parameters( price_in_cents: int, # this is the money which you, as the seller, will receive session_id: str, ) -> dict[str, str]: - market_sell_parameters = {} - - market_sell_parameters["sessionid"] = session_id - market_sell_parameters["appid"] = "753" - market_sell_parameters["contextid"] = "6" - market_sell_parameters["assetid"] = asset_id # To automatically determine asset ID, use retrieve_asset_id(). - market_sell_parameters["amount"] = "1" - market_sell_parameters["price"] = str(price_in_cents) - - return market_sell_parameters + return { + "sessionid": session_id, + "appid": "753", + "contextid": "6", + "assetid": asset_id, + "amount": "1", + "price": str(price_in_cents), + } def get_request_headers() -> dict[str, str]: @@ -238,6 +237,7 @@ def get_request_headers() -> dict[str, str]: def sell_booster_pack( asset_id: str, price_in_cents: int, # this is the money which you, as the seller, will receive + *, verbose: bool = True, ) -> dict | None: cookie = get_cookie_dict() @@ -300,6 +300,7 @@ def sell_booster_pack( def retrieve_asset_id( listing_hash: str, steam_inventory: dict | None = None, + *, focus_on_marketable_items: bool = True, profile_id: str | None = None, verbose: bool = True, @@ -377,6 +378,7 @@ def create_booster_packs_for_batch(listing_hashes: list[str]) -> dict[str, dict def sell_booster_packs_for_batch( price_dict_for_listing_hashes: dict[str, int], + *, update_steam_inventory: bool = True, focus_on_marketable_items: bool = True, profile_id: str | None = None, @@ -406,6 +408,7 @@ def sell_booster_packs_for_batch( def create_then_sell_booster_packs_for_batch( price_dict_for_listing_hashes: dict[str, int], + *, update_steam_inventory: bool = True, focus_on_marketable_items: bool = True, profile_id: str | None = None, @@ -428,6 +431,7 @@ def create_then_sell_booster_packs_for_batch( def update_and_save_next_creation_times( creation_results: dict[str, dict | None], + *, verbose: bool = True, next_creation_time_file_name: str | None = None, ) -> dict[str, str]: @@ -444,9 +448,7 @@ def update_and_save_next_creation_times( save_to_disk = False is_first_displayed_line = True - for listing_hash in creation_results: - result = creation_results[listing_hash] - + for listing_hash, result in creation_results.items(): if result is not None: app_id = convert_listing_hash_to_app_id(listing_hash) next_creation_times[app_id] = formatted_next_creation_time @@ -476,7 +478,7 @@ def main() -> None: price_dict_for_listing_hashes = {listing_hash: price_in_cents} - creation_results, sale_results = create_then_sell_booster_packs_for_batch( + _creation_results, _sale_results = create_then_sell_booster_packs_for_batch( price_dict_for_listing_hashes, ) diff --git a/src/json_utils.py b/src/json_utils.py index 65840b3..0dd35ac 100644 --- a/src/json_utils.py +++ b/src/json_utils.py @@ -7,7 +7,13 @@ def load_json(fname: str) -> dict: return json.load(f) -def save_json(data: dict, fname: str, prettify: bool = True, indent: int = 4) -> None: +def save_json( + data: dict, + fname: str, + *, + prettify: bool = True, + indent: int = 4, +) -> None: with Path(fname).open("w", encoding="utf8") as f: if prettify: json.dump(data, f, indent=indent) diff --git a/src/list_possible_lures.py b/src/list_possible_lures.py index d366330..c2a490f 100644 --- a/src/list_possible_lures.py +++ b/src/list_possible_lures.py @@ -77,6 +77,7 @@ def filter_app_ids_with_potential_profit( sell_prices_without_fee: dict[str, float], gem_amounts_for_a_booster_pack: dict[str, int], gem_sack_price_in_euros: float | None = None, + *, verbose: bool = True, ) -> list[str]: # Filter out appIDs for which the sell price (without fee) is lower than the cost to craft a Booster Pack. @@ -117,6 +118,7 @@ def filter_app_ids_with_potential_profit( def remove_app_ids_previously_processed( filtered_app_ids: list[str], app_ids_previously_processed: list[str] | None = None, + *, verbose: bool = True, ) -> list[str]: # Manually remove previously processed appIDs from the list of returned appIDs of interest. diff --git a/src/market_arbitrage_utils.py b/src/market_arbitrage_utils.py index 566ea68..1fdfb70 100644 --- a/src/market_arbitrage_utils.py +++ b/src/market_arbitrage_utils.py @@ -16,35 +16,27 @@ def determine_whether_booster_pack_was_crafted_at_least_once(badge_data: dict) -> bool: - next_creation_time = badge_data["next_creation_time"] - - return bool(next_creation_time is not None) + return badge_data["next_creation_time"] is not None def filter_out_badges_never_crafted( aggregated_badge_data: dict[str, dict], + *, verbose: bool = True, ) -> dict[str, dict]: # Filter out games for which a booster pack was never crafted (according to 'data/next_creation_times.json'), # thus focus on games which are tracked more closely, because they are likely to show a market arbitrage (again). - - filtered_badge_data = {} - - for app_id in aggregated_badge_data: - individual_badge_data = aggregated_badge_data[app_id] - - booster_pack_is_tracked = ( - determine_whether_booster_pack_was_crafted_at_least_once( - individual_badge_data, - ) - ) - - if booster_pack_is_tracked: - filtered_badge_data[app_id] = individual_badge_data + filtered_badge_data = { + app_id: data + for app_id, data in aggregated_badge_data.items() + if determine_whether_booster_pack_was_crafted_at_least_once(data) + } if verbose: print( - f"There are {len(filtered_badge_data)} booster packs which are tracked, as they were crafted at least once. ({len(aggregated_badge_data) - len(filtered_badge_data)} omitted)", + f"There are {len(filtered_badge_data)} booster packs which are tracked, " + f"as they were crafted at least once. " + f"({len(aggregated_badge_data) - len(filtered_badge_data)} omitted)", ) return filtered_badge_data @@ -52,6 +44,7 @@ def filter_out_badges_never_crafted( def filter_out_badges_recently_crafted( aggregated_badge_data: dict[str, dict], + *, verbose: bool = True, ) -> dict[str, dict]: # Filter out games for which a booster pack was crafted less than 24 hours ago, @@ -61,9 +54,7 @@ def filter_out_badges_recently_crafted( current_time = get_current_time() - for app_id in aggregated_badge_data: - individual_badge_data = aggregated_badge_data[app_id] - + for app_id, individual_badge_data in aggregated_badge_data.items(): booster_pack_can_be_crafted = determine_whether_a_booster_pack_can_be_crafted( individual_badge_data, current_time, @@ -74,7 +65,8 @@ def filter_out_badges_recently_crafted( if verbose: print( - f"There are {len(filtered_badge_data)} booster packs which can be immediately crafted. ({len(aggregated_badge_data) - len(filtered_badge_data)} excluded because of cooldown)", + f"There are {len(filtered_badge_data)} booster packs which can be immediately crafted. " + f"({len(aggregated_badge_data) - len(filtered_badge_data)} excluded because of cooldown)", ) return filtered_badge_data @@ -124,6 +116,7 @@ def filter_out_badges_with_low_sell_price( aggregated_badge_data: dict[str, dict], user_chosen_price_threshold: float | None = None, category_name: str | None = None, + *, verbose: bool = True, ) -> dict[str, dict]: # Filter out games for which the sell price (ask) is lower than the gem price, @@ -143,9 +136,7 @@ def filter_out_badges_with_low_sell_price( unknown_price_counter = 0 - for app_id in aggregated_badge_data: - individual_badge_data = aggregated_badge_data[app_id] - + for app_id, individual_badge_data in aggregated_badge_data.items(): sell_price_is_unknown = determine_whether_sell_price_is_unknown( individual_badge_data, ) @@ -172,6 +163,7 @@ def filter_out_badges_with_low_sell_price( def find_badge_arbitrages( badge_data: dict, market_order_dict: dict[str, dict] | None = None, + *, verbose: bool = False, ) -> dict[str, dict]: if market_order_dict is None: @@ -183,9 +175,7 @@ def find_badge_arbitrages( badge_arbitrages: dict[str, dict] = {} - for app_id in badge_data: - individual_badge_data = badge_data[app_id] - + for app_id, individual_badge_data in badge_data.items(): gem_price_including_fee = individual_badge_data["gem_price"] listing_hash = individual_badge_data["listing_hash"] @@ -264,6 +254,7 @@ def find_badge_arbitrages( def print_arbitrages( badge_arbitrages: dict[str, dict], + *, use_numbered_bullet_points: bool = False, use_hyperlink: bool = False, ) -> None: @@ -316,6 +307,7 @@ def print_arbitrages( def convert_arbitrages_for_batch_create_then_sell( badge_arbitrages: dict[str, dict], profit_threshold: float = 0.01, # profit in euros + *, verbose: bool = True, ) -> dict[str, int]: # Code inspired from print_arbitrages() @@ -348,6 +340,7 @@ def convert_arbitrages_for_batch_create_then_sell( def update_badge_arbitrages_with_latest_market_order_data( badge_data: dict[str, dict], arbitrage_data: dict[str, dict], + *, retrieve_market_orders_online: bool = True, verbose: bool = False, ) -> dict[str, dict]: @@ -356,9 +349,7 @@ def update_badge_arbitrages_with_latest_market_order_data( # Based on arbitrage_data, select the badge_data for which we want to download (again) the latest market orders: selected_badge_data = {} - for listing_hash in arbitrage_data: - arbitrage = arbitrage_data[listing_hash] - + for listing_hash, arbitrage in arbitrage_data.items(): if arbitrage["is_marketable"] and arbitrage["profit"] > 0: app_id = convert_listing_hash_to_app_id(listing_hash) selected_badge_data[app_id] = badge_data[app_id] @@ -377,6 +368,7 @@ def update_badge_arbitrages_with_latest_market_order_data( def get_filtered_badge_data( + *, retrieve_listings_from_scratch: bool = True, enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, @@ -385,7 +377,7 @@ def get_filtered_badge_data( from_javascript: bool = False, ) -> dict[str, dict]: aggregated_badge_data = load_aggregated_badge_data( - retrieve_listings_from_scratch, + retrieve_listings_from_scratch=retrieve_listings_from_scratch, enforced_sack_of_gems_price=enforced_sack_of_gems_price, minimum_allowed_sack_of_gems_price=minimum_allowed_sack_of_gems_price, from_javascript=from_javascript, diff --git a/src/market_buzz_utils.py b/src/market_buzz_utils.py index 6f6423c..67e8ea3 100644 --- a/src/market_buzz_utils.py +++ b/src/market_buzz_utils.py @@ -13,9 +13,10 @@ def filter_listings( all_listings: dict[str, dict] | None = None, - min_sell_price: float = 30, # in cents + min_sell_price: int = 30, # in cents min_num_listings: int = 20, # to remove listings with very few sellers, who chose unrealistic sell prices + *, verbose: bool = True, ) -> list[str]: if all_listings is None: @@ -69,17 +70,17 @@ def filter_out_unmarketable_packs( marketable_market_order_dict = {} unknown_market_order_dict = {} - for listing_hash in market_order_dict: + for listing_hash, current_data in market_order_dict.items(): try: - is_marketable = market_order_dict[listing_hash]["is_marketable"] + is_marketable = current_data["is_marketable"] except KeyError: print(f"Marketable status not found for {listing_hash}") - unknown_market_order_dict[listing_hash] = market_order_dict[listing_hash] + unknown_market_order_dict[listing_hash] = current_data is_marketable = False # avoid taking any risk: ASSUME the booster pack is NOT marketable if is_marketable: - marketable_market_order_dict[listing_hash] = market_order_dict[listing_hash] + marketable_market_order_dict[listing_hash] = current_data return marketable_market_order_dict, unknown_market_order_dict @@ -91,7 +92,7 @@ def sort_according_to_buzz( if marketable_market_order_dict is None: ( marketable_market_order_dict, - unknown_market_order_dict, + _unknown_market_order_dict, ) = filter_out_unmarketable_packs(market_order_dict) return sorted( @@ -157,6 +158,7 @@ def print_packs_with_high_buzz( def fill_in_badge_data_with_data_from_steam_card_exchange( all_listings: dict[str, dict], aggregated_badge_data: dict[str, dict] | None = None, + *, force_update_from_steam_card_exchange: bool = False, enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, diff --git a/src/market_foil_utils.py b/src/market_foil_utils.py index 8b09096..de90716 100644 --- a/src/market_foil_utils.py +++ b/src/market_foil_utils.py @@ -31,6 +31,7 @@ def get_item_type_no_for_trading_cards( listing_hash: str | None = None, all_listing_details: dict[str, dict] | None = None, listing_details_output_file_name: str | None = None, + *, verbose: bool = True, ) -> int: # Caveat: the item type is not always equal to 2. Check appID 232770 (POSTAL) for example! @@ -104,7 +105,7 @@ def get_item_type_no_for_trading_cards( return item_type_no -def get_border_color_no_for_trading_cards(is_foil: bool = False) -> int: +def get_border_color_no_for_trading_cards(*, is_foil: bool = False) -> int: # NB: this leads to a goo value 10 times higher than with border_corlor_no equal to zero. However, it seems to # be applied without any check, so that the returned goo values are misleading when applied to any item other # than a trading card, such as an emoticon and a profile background. @@ -115,6 +116,7 @@ def get_steam_goo_value_parameters( app_id: str, item_type: int | None = None, listing_hash: str | None = None, + *, is_foil: bool = True, verbose: bool = True, ) -> dict[str, str | int]: @@ -126,11 +128,11 @@ def get_steam_goo_value_parameters( border_color = get_border_color_no_for_trading_cards(is_foil=is_foil) - params: dict[str, str | int] = {} - - params["appid"] = app_id - params["item_type"] = item_type - params["border_color"] = border_color + params: dict[str, str | int] = { + "appid": app_id, + "item_type": item_type, + "border_color": border_color, + } return params @@ -138,6 +140,7 @@ def get_steam_goo_value_parameters( def query_goo_value( app_id: str, item_type: int | None, + *, verbose: bool = True, ) -> int | None: cookie = get_cookie_dict() @@ -180,6 +183,7 @@ def query_goo_value( def get_listings_for_foil_cards( + *, retrieve_listings_from_scratch: bool, listing_output_file_name: str | None = None, start_index: int = 0, @@ -201,6 +205,7 @@ def get_listings_for_foil_cards( def group_listing_hashes_by_app_id( all_listings: dict[str, dict], + *, verbose: bool = True, ) -> dict[str, list[str]]: groups_by_app_id: dict[str, list[str]] = {} @@ -224,9 +229,7 @@ def find_cheapest_listing_hashes( ) -> list[str]: cheapest_listing_hashes = [] - for app_id in groups_by_app_id: - listing_hashes = groups_by_app_id[app_id] - + for listing_hashes in groups_by_app_id.values(): # Sort with respect to two attributes: # - ascending sell prices, # - **descending** volumes. @@ -253,7 +256,7 @@ def find_representative_listing_hashes( ) -> list[str]: representative_listing_hashes = [] - for app_id in groups_by_app_id: + for app_id, listing_hashes in groups_by_app_id.items(): if dictionary_of_representative_listing_hashes is not None: try: # For retro-compatibility, we try to use representative for which we previously downloaded item name ids @@ -265,8 +268,6 @@ def find_representative_listing_hashes( else: previously_used_listing_hashes_for_app_id = None - listing_hashes = groups_by_app_id[app_id] - # Sort with respect to lexicographical order. sorted_listing_hashes = sorted(listing_hashes) @@ -301,6 +302,7 @@ def filter_listings_with_arbitrary_price_threshold( all_listings: dict[str, dict], listing_hashes_to_filter_from: list[str], price_threshold_in_cents: float | None = None, + *, verbose: bool = True, ) -> list[str]: if price_threshold_in_cents is not None: @@ -323,6 +325,7 @@ def filter_listings_with_arbitrary_price_threshold( def load_all_goo_details( goo_details_file_name: str | None = None, + *, verbose: bool = True, ) -> dict[str, int | None]: if goo_details_file_name is None: @@ -351,6 +354,7 @@ def save_all_goo_details( def filter_out_listing_hashes_if_goo_details_are_already_known_for_app_id( filtered_cheapest_listing_hashes: list[str], goo_details_file_name_for_for_foil_cards: str | None = None, + *, verbose: bool = True, ) -> list[str]: # Filter out listings associated with an appID for which we already know the goo details. @@ -440,6 +444,7 @@ def get_minimal_ask_price_in_euros_on_steam_market() -> float: def compute_unrewarding_threshold_in_gems( sack_of_gems_price_in_euros: float | None = None, + *, retrieve_gem_price_from_scratch: bool = False, verbose: bool = True, ) -> float: @@ -470,6 +475,7 @@ def discard_necessarily_unrewarding_app_ids( app_ids_with_unreliable_goo_details: list[str] | None = None, app_ids_with_unknown_goo_value: list[str] | None = None, sack_of_gems_price_in_euros: float | None = None, + *, retrieve_gem_price_from_scratch: bool = False, verbose: bool = True, ) -> list[str]: @@ -491,9 +497,7 @@ def discard_necessarily_unrewarding_app_ids( potentially_rewarding_app_ids = [] - for app_id in all_goo_details: - goo_value_in_gems = all_goo_details[app_id] - + for app_id, goo_value_in_gems in all_goo_details.items(): if app_id in app_ids_to_omit: continue @@ -531,6 +535,7 @@ def find_listing_hashes_with_unknown_goo_value( listing_candidates: list[str], app_ids_with_unreliable_goo_details: list[str], all_goo_details: dict[str, int | None], + *, verbose: bool = True, ) -> list[str]: app_ids_with_unknown_goo_value = [] @@ -565,6 +570,7 @@ def determine_whether_an_arbitrage_might_exist_for_foil_cards( all_listings: dict[str, dict] | None = None, listing_output_file_name: str | None = None, sack_of_gems_price_in_euros: float | None = None, + *, retrieve_gem_price_from_scratch: bool = True, verbose: bool = True, ) -> dict[str, dict[str, float]]: @@ -634,11 +640,12 @@ def determine_whether_an_arbitrage_might_exist_for_foil_cards( is_arbitrage = bool(profit_in_cents > 0) if is_arbitrage: - arbitrage = {} - arbitrage["profit"] = profit_in_cents / 100 - arbitrage["ask"] = ask_in_cents / 100 - arbitrage["goo_amount"] = goo_value_in_gems - arbitrage["goo_value"] = goo_value_in_cents / 100 + arbitrage = { + "profit": profit_in_cents / 100, + "ask": ask_in_cents / 100, + "goo_amount": goo_value_in_gems, + "goo_value": goo_value_in_cents / 100, + } arbitrages[listing_hash] = arbitrage @@ -647,6 +654,7 @@ def determine_whether_an_arbitrage_might_exist_for_foil_cards( def print_arbitrages_for_foil_cards( arbitrages: dict[str, dict[str, float]], + *, use_numbered_bullet_points: bool = False, ) -> None: bullet_point = get_bullet_point_for_display( @@ -697,6 +705,7 @@ def find_app_ids_with_unknown_item_type_for_their_representatives( listing_candidates: list[str], all_listing_details: dict[str, dict] | None = None, listing_details_output_file_name: str | None = None, + *, verbose: bool = True, ) -> list[str]: dictionary_of_representative_listing_hashes = ( @@ -736,6 +745,7 @@ def download_missing_goo_details( goo_details_file_name_for_for_foil_cards: str | None = None, enforced_app_ids_to_process: list[str] | None = None, num_queries_between_save: int = 100, + *, verbose: bool = True, ) -> dict[str, int | None]: if goo_details_file_name_for_for_foil_cards is None: @@ -863,6 +873,7 @@ def find_item_type_for_app_id( all_listing_details: dict[str, dict] | None = None, listing_details_output_file_name: str | None = None, dictionary_of_representative_listing_hashes: dict[str, list[str]] | None = None, + *, verbose: bool = False, ) -> int | None: if listing_details_output_file_name is None: @@ -903,6 +914,7 @@ def download_goo_value_for_app_id( all_listing_details: dict[str, dict] | None = None, listing_details_output_file_name: str | None = None, dictionary_of_representative_listing_hashes: dict[str, list[str]] | None = None, + *, verbose: bool = True, ) -> int | None: item_type = find_item_type_for_app_id( diff --git a/src/market_gamble_utils.py b/src/market_gamble_utils.py index 897562e..7de29f6 100644 --- a/src/market_gamble_utils.py +++ b/src/market_gamble_utils.py @@ -1,5 +1,6 @@ import time +from src.api_utils import get_rate_limits from src.drop_rate_estimates import ( clamp_proportion, get_drop_rate_estimates_based_on_item_rarity_pattern, @@ -12,7 +13,6 @@ load_market_order_data_from_disk, ) from src.market_search import ( - get_steam_api_rate_limits_for_market_search, get_tag_item_class_no_for_emoticons, get_tag_item_class_no_for_profile_backgrounds, get_tag_item_class_no_for_trading_cards, @@ -29,6 +29,9 @@ get_listing_output_file_name_for_profile_backgrounds, ) +type ItemRarityPattern = tuple[int, int, int] +type DropRateEstimates = dict[ItemRarityPattern, float] + def update_all_listings_for_foil_cards(start_index: int = 0) -> None: print("Downloading listings for foil cards.") @@ -94,7 +97,10 @@ def update_all_listings_for_items_other_than_cards( cookie = get_cookie_dict() has_secured_cookie = bool(len(cookie) > 0) - rate_limits = get_steam_api_rate_limits_for_market_search(has_secured_cookie) + rate_limits = get_rate_limits( + "market_search", + has_secured_cookie=has_secured_cookie, + ) cooldown_duration = rate_limits["cooldown"] print( @@ -112,6 +118,7 @@ def update_all_listings_for_items_other_than_cards( def get_listings( listing_output_file_name: str, + *, retrieve_listings_from_scratch: bool = False, ) -> dict[str, dict]: if retrieve_listings_from_scratch: @@ -126,9 +133,9 @@ def filter_out_candidates_whose_ask_price_is_below_threshold( item_rarity_patterns_per_app_id: dict[str, dict], price_threshold_in_cents: float | None = None, category_name: str | None = None, - drop_rate_estimates_for_common_rarity: dict[tuple[int, int, int], float] - | None = None, + drop_rate_estimates_for_common_rarity: DropRateEstimates | None = None, gem_price_in_euros: float | None = None, + *, verbose: bool = True, ) -> dict[str, dict]: if gem_price_in_euros is None: @@ -207,6 +214,7 @@ def filter_out_candidates_whose_ask_price_is_below_threshold( def get_market_orders( filtered_badge_data: dict[str, dict], + *, retrieve_market_orders_online: bool, focus_on_listing_hashes_never_seen_before: bool, listing_details_output_file_name: str, @@ -223,13 +231,11 @@ def get_market_orders( # Filter out listing hashes which have already been encountered at least once - first_encountered_filtered_badge_data = {} - - for dummy_app_id in filtered_badge_data: - if filtered_badge_data[dummy_app_id]["listing_hash"] not in market_order_dict: - first_encountered_filtered_badge_data[dummy_app_id] = filtered_badge_data[ - dummy_app_id - ] + first_encountered_filtered_badge_data = { + dummy_app_id: dummy_data + for dummy_app_id, dummy_data in filtered_badge_data.items() + if dummy_data["listing_hash"] not in market_order_dict + } # Retrieval of market orders (bid, ask) @@ -294,6 +300,7 @@ def count_listing_hashes_per_app_id(all_listings: dict[str, dict]) -> dict[str, def get_listings_with_other_rarity_tags( + *, look_for_profile_backgrounds: bool, retrieve_listings_with_another_rarity_tag_from_scratch: bool = False, ) -> tuple[dict[str, dict], dict[str, dict]]: diff --git a/src/market_listing.py b/src/market_listing.py index 4d1ff89..53ab774 100644 --- a/src/market_listing.py +++ b/src/market_listing.py @@ -6,6 +6,7 @@ import requests from bs4 import BeautifulSoup +from src.api_utils import get_rate_limits from src.json_utils import load_json, save_json from src.market_search import load_all_listings from src.personal_info import ( @@ -14,7 +15,6 @@ ) from src.utils import ( TIMEOUT_IN_SECONDS, - get_cushioned_cooldown_in_seconds, get_listing_details_output_file_name, ) @@ -22,6 +22,7 @@ def get_steam_market_listing_url( app_id: str | None = None, listing_hash: str | None = None, + *, render_as_json: bool = True, replace_spaces: bool = False, replace_parenthesis: bool = False, @@ -60,26 +61,6 @@ def get_listing_parameters() -> dict[str, str]: return {"currency": "3"} -def get_steam_api_rate_limits_for_market_listing( - has_secured_cookie: bool = False, -) -> dict[str, int]: - # Objective: return the rate limits of Steam API for the market. - - if has_secured_cookie: - rate_limits = { - "max_num_queries": 25, - "cooldown": get_cushioned_cooldown_in_seconds(num_minutes=3), - } - - else: - rate_limits = { - "max_num_queries": 25, - "cooldown": get_cushioned_cooldown_in_seconds(num_minutes=5), - } - - return rate_limits - - def figure_out_relevant_id( asset_dict: dict[str, dict], asset_ids: list[str], @@ -265,6 +246,7 @@ def parse_item_name_id(html_doc: str) -> tuple[int | None, bool | None, int | No def get_listing_details( listing_hash: str, cookie: dict[str, str] | None = None, + *, render_as_json: bool = False, ) -> tuple[dict[str, dict], int]: listing_details: dict[str, dict] = {} @@ -308,10 +290,11 @@ def get_listing_details( if item_type_no is None: print(f"Item type not found for {listing_hash}") - listing_details[listing_hash] = {} - listing_details[listing_hash]["item_nameid"] = item_nameid - listing_details[listing_hash]["is_marketable"] = is_marketable - listing_details[listing_hash]["item_type_no"] = item_type_no + listing_details[listing_hash] = { + "item_nameid": item_nameid, + "is_marketable": is_marketable, + "item_type_no": item_type_no, + } status_code = resp_data.status_code return listing_details, status_code @@ -320,6 +303,7 @@ def get_listing_details( def get_listing_details_batch( listing_hashes: list[str] | dict[str, dict], all_listing_details: dict[str, dict] | None = None, + *, save_to_disk: bool = True, listing_details_output_file_name: str | None = None, ) -> dict[str, dict]: @@ -329,7 +313,10 @@ def get_listing_details_batch( cookie = get_cookie_dict() has_secured_cookie = bool(len(cookie) > 0) - rate_limits = get_steam_api_rate_limits_for_market_listing(has_secured_cookie) + rate_limits = get_rate_limits( + "market_listing", + has_secured_cookie=has_secured_cookie, + ) if all_listing_details is None: all_listing_details = {} @@ -339,7 +326,7 @@ def get_listing_details_batch( query_count = 0 for count, listing_hash in enumerate(listing_hashes): - if count + 1 % 100 == 0: + if (count + 1) % 100 == 0: print(f"[{count + 1}/{num_listings}]") listing_details, status_code = get_listing_details( diff --git a/src/market_order.py b/src/market_order.py index 683f863..334a7a7 100644 --- a/src/market_order.py +++ b/src/market_order.py @@ -4,10 +4,12 @@ from contextlib import suppress from datetime import timedelta from http import HTTPStatus +from typing import Final import requests from requests.exceptions import ConnectionError, ReadTimeout +from src.api_utils import INTER_REQUEST_COOLDOWN_FIELD, get_rate_limits from src.cookie_utils import force_update_sessionid from src.creation_time_utils import get_current_time, to_timestamp from src.json_utils import load_json, save_json @@ -18,14 +20,13 @@ ) from src.utils import ( TIMEOUT_IN_SECONDS, - get_cushioned_cooldown_in_seconds, get_market_order_file_name, ) -INTER_REQUEST_COOLDOWN_FIELD = "cooldown_between_each_request" +type MarketOrderData = dict[str, float | int | bool] -UPDATE_COOLDOWN_FIELD = "update_timestamp" -UPDATE_COOLDOWN_IN_HOURS = 72 +UPDATE_COOLDOWN_FIELD: Final[str] = "update_timestamp" +UPDATE_COOLDOWN_IN_HOURS: Final[int] = 72 def get_steam_market_order_url() -> str: @@ -33,37 +34,13 @@ def get_steam_market_order_url() -> str: def get_market_order_parameters(item_nameid: str) -> dict[str, str]: - params = {} - - params["country"] = "FR" - params["language"] = "english" - params["currency"] = "3" - params["item_nameid"] = item_nameid - params["two_factor"] = "0" - - return params - - -def get_steam_api_rate_limits_for_market_order( - has_secured_cookie: bool = False, -) -> dict[str, int]: - # Objective: return the rate limits of Steam API for the market. - - if has_secured_cookie: - rate_limits = { - "max_num_queries": 50, - "cooldown": get_cushioned_cooldown_in_seconds(num_minutes=1), - } - - else: - rate_limits = { - "max_num_queries": 25, - "cooldown": get_cushioned_cooldown_in_seconds(num_minutes=5), - } - - rate_limits[INTER_REQUEST_COOLDOWN_FIELD] = 0 - - return rate_limits + return { + "country": "FR", + "language": "english", + "currency": "3", + "item_nameid": item_nameid, + "two_factor": "0", + } def get_market_order_headers() -> dict[str, str]: @@ -86,6 +63,7 @@ def get_market_order_headers() -> dict[str, str]: def download_market_order_data( listing_hash: str, item_nameid: str | None = None, + *, verbose: bool = False, listing_details_output_file_name: str | None = None, ) -> tuple[float, float, int, int]: @@ -194,7 +172,7 @@ def download_market_order_data( def is_dummy_market_order_data( - market_order_data: dict[str, float | int | bool], + market_order_data: MarketOrderData, ) -> bool: bid_price = market_order_data["bid"] ask_price = market_order_data["ask"] @@ -205,7 +183,7 @@ def is_dummy_market_order_data( def has_a_recent_timestamp( - market_order_data: dict[str, float | int | bool], + market_order_data: MarketOrderData, threshold_timestamp: int, ) -> bool: last_update_timestamp = market_order_data[UPDATE_COOLDOWN_FIELD] @@ -216,6 +194,7 @@ def has_a_recent_timestamp( def download_market_order_data_batch( badge_data: dict[str, dict], market_order_dict: dict[str, dict] | None = None, + *, verbose: bool = False, save_to_disk: bool = True, market_order_output_file_name: str | None = None, @@ -241,7 +220,7 @@ def download_market_order_data_batch( cookie = force_update_sessionid(cookie) has_secured_cookie = bool(len(cookie) > 0) - rate_limits = get_steam_api_rate_limits_for_market_order(has_secured_cookie) + rate_limits = get_rate_limits("market_order", has_secured_cookie=has_secured_cookie) if market_order_dict is None: market_order_dict = {} @@ -316,6 +295,7 @@ def download_market_order_data_batch( def load_market_order_data( badge_data: dict[str, dict], + *, trim_output: bool = False, retrieve_market_orders_online: bool = True, verbose: bool = False, @@ -391,7 +371,7 @@ def main() -> bool: # Download based on a listing hash - bid_price, ask_price, bid_volume, ask_volume = download_market_order_data( + _bid_price, _ask_price, _bid_volume, _ask_volume = download_market_order_data( listing_hash, verbose=True, ) @@ -400,8 +380,7 @@ def main() -> bool: app_id = listing_hash.split("-", maxsplit=1)[0] - badge_data: dict[str, dict] = {} - badge_data[app_id] = {} + badge_data: dict[str, dict] = {app_id: {}} badge_data[app_id]["listing_hash"] = listing_hash download_market_order_data_batch( @@ -421,7 +400,7 @@ def main() -> bool: "505730-Holy Potatoes! We’re in Space%3F! Booster Pack", ] for listing_hash_to_test in listing_hashes: - bid_price, ask_price, bid_volume, ask_volume = download_market_order_data( + _bid_price, _ask_price, _bid_volume, _ask_volume = download_market_order_data( listing_hash_to_test, verbose=True, ) diff --git a/src/market_search.py b/src/market_search.py index 2d49be4..92f81a5 100644 --- a/src/market_search.py +++ b/src/market_search.py @@ -7,14 +7,15 @@ import requests from requests.exceptions import ConnectionError +from src.api_utils import get_rate_limits from src.json_utils import load_json, save_json from src.personal_info import ( get_cookie_dict, update_and_save_cookie_to_disk_if_values_changed, ) +from src.tag_utils import get_tag_drop_rate_str from src.utils import ( TIMEOUT_IN_SECONDS, - get_cushioned_cooldown_in_seconds, get_listing_output_file_name, ) @@ -39,29 +40,13 @@ def get_tag_item_class_no_for_booster_packs() -> int: return 5 -def get_tag_drop_rate_str(rarity: str | None = None) -> str: - if rarity is None: - rarity = "common" - - if rarity == "extraordinary": - tag_drop_rate_no = 3 - elif rarity == "rare": - tag_drop_rate_no = 2 - elif rarity == "uncommon": - tag_drop_rate_no = 1 - else: - # Rarity: Common - tag_drop_rate_no = 0 - - return f"tag_droprate_{tag_drop_rate_no}" - - def get_search_parameters( start_index: int = 0, delta_index: int = 100, tag_item_class_no: int | None = None, tag_drop_rate_str: str | None = None, rarity: str | None = None, + *, is_foil_trading_card: bool = True, ) -> dict[str, str]: if tag_drop_rate_str is None: @@ -81,47 +66,26 @@ def get_search_parameters( column_to_sort_by = "name" sort_direction = "asc" - params = {} - - params["norender"] = "1" - params["category_753_Game[]"] = "any" - params["category_753_droprate[]"] = tag_drop_rate_str - params["category_753_item_class[]"] = f"tag_item_class_{tag_item_class_no}" - params["appid"] = "753" - params["sort_column"] = column_to_sort_by - params["sort_dir"] = sort_direction - params["start"] = str(start_index) - params["count"] = str(delta_index) + params = { + "norender": "1", + "category_753_Game[]": "any", + "category_753_droprate[]": tag_drop_rate_str, + "category_753_item_class[]": f"tag_item_class_{tag_item_class_no}", + "appid": "753", + "sort_column": column_to_sort_by, + "sort_dir": sort_direction, + "start": str(start_index), + "count": str(delta_index), + } if tag_item_class_no == get_tag_item_class_no_for_trading_cards(): - if is_foil_trading_card: - params["category_753_cardborder[]"] = "tag_cardborder_1" - else: - params["category_753_cardborder[]"] = "tag_cardborder_0" + params["category_753_cardborder[]"] = ( + "tag_cardborder_1" if is_foil_trading_card else "tag_cardborder_0" + ) return params -def get_steam_api_rate_limits_for_market_search( - has_secured_cookie: bool = False, -) -> dict[str, int]: - # Objective: return the rate limits of Steam API for the market. - - if has_secured_cookie: - rate_limits = { - "max_num_queries": 50, - "cooldown": get_cushioned_cooldown_in_seconds(num_minutes=1), - } - - else: - rate_limits = { - "max_num_queries": 25, - "cooldown": get_cushioned_cooldown_in_seconds(num_minutes=5), - } - - return rate_limits - - def get_all_listings( all_listings: dict[str, dict] | None = None, url: str | None = None, @@ -137,7 +101,10 @@ def get_all_listings( cookie = get_cookie_dict() has_secured_cookie = bool(len(cookie) > 0) - rate_limits = get_steam_api_rate_limits_for_market_search(has_secured_cookie) + rate_limits = get_rate_limits( + "market_search", + has_secured_cookie=has_secured_cookie, + ) if all_listings is None: all_listings = {} @@ -211,11 +178,11 @@ def get_all_listings( listings: dict[str, dict] = {} for listing in result["results"]: listing_hash = listing["hash_name"] - - listings[listing_hash] = {} - listings[listing_hash]["sell_listings"] = listing["sell_listings"] - listings[listing_hash]["sell_price"] = listing["sell_price"] - listings[listing_hash]["sell_price_text"] = listing["sell_price_text"] + listings[listing_hash] = { + "sell_listings": listing["sell_listings"], + "sell_price": listing["sell_price"], + "sell_price_text": listing["sell_price_text"], + } else: status_code = resp_data.status_code if resp_data else None diff --git a/src/market_utils.py b/src/market_utils.py index 412bd7d..3e9cc9e 100644 --- a/src/market_utils.py +++ b/src/market_utils.py @@ -17,6 +17,7 @@ def determine_whether_listing_hash_is_dubious(listing_hash: str) -> bool: def filter_out_dubious_listing_hashes( all_listings: dict[str, dict], + *, verbose: bool = True, ) -> dict[str, dict]: # Filter out listing hashes which hint at a dubious market listing for the booster pack. For instance: @@ -25,9 +26,7 @@ def filter_out_dubious_listing_hashes( filtered_listings = {} - for listing_hash in all_listings: - individual_market_listing = all_listings[listing_hash] - + for listing_hash, individual_market_listing in all_listings.items(): booster_pack_is_dubious = determine_whether_listing_hash_is_dubious( listing_hash, ) @@ -48,6 +47,7 @@ def filter_out_dubious_listing_hashes( def match_badges_with_listing_hashes( badge_creation_details: dict[str, dict] | None = None, all_listings: dict[str, dict] | None = None, + *, verbose: bool = True, ) -> dict[str, str | None]: # Badges for games which I own @@ -109,6 +109,7 @@ def aggregate_badge_data( all_listings: dict[str, dict] | None = None, enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, + *, retrieve_gem_price_from_scratch: bool = False, ) -> dict[str, dict]: # Aggregate data: @@ -171,6 +172,7 @@ def aggregate_badge_data( def load_aggregated_badge_data( + *, retrieve_listings_from_scratch: bool = False, enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, @@ -225,7 +227,7 @@ def populate_random_samples_of_badge_data( return True -def main(populate_all_item_name_ids: bool = False) -> bool: +def main(*, populate_all_item_name_ids: bool = False) -> bool: if populate_all_item_name_ids: # Pre-retrieval of ALL the MISSING item name ids. # Caveat: this may require a long time, due to API rate limits. diff --git a/src/parsing_utils.py b/src/parsing_utils.py index 6460159..2bbeb05 100644 --- a/src/parsing_utils.py +++ b/src/parsing_utils.py @@ -13,6 +13,7 @@ def parse_javascript_one_liner( badges_as_str: str, + *, verbose: bool = False, ) -> dict[str, dict]: badge_creation_details = {} @@ -44,6 +45,7 @@ def parse_javascript_one_liner( def parse_augmented_steam_drop_down_menu( lines: list[str], + *, verbose: bool = False, ) -> dict[str, dict]: badge_creation_details: dict[str, dict] = {} @@ -75,6 +77,7 @@ def parse_augmented_steam_drop_down_menu( def parse_badge_creation_details( badge_creation_file_name: str | None = None, + *, from_javascript: bool = False, verbose: bool = False, ) -> dict[str, dict]: diff --git a/src/personal_info.py b/src/personal_info.py index 96b13de..e556576 100644 --- a/src/personal_info.py +++ b/src/personal_info.py @@ -36,7 +36,7 @@ def save_steam_cookie_to_disk( return is_cookie_to_be_saved -def get_cookie_dict(verbose: bool = False) -> dict[str, str]: +def get_cookie_dict(*, verbose: bool = False) -> dict[str, str]: cookie = load_steam_cookie_from_disk() if verbose: @@ -49,16 +49,16 @@ def get_cookie_dict(verbose: bool = False) -> dict[str, str]: def update_cookie_dict( original_cookie: dict[str, str], dict_with_new_values: dict[str, str], + *, verbose: bool = False, ) -> dict[str, str]: cookie = original_cookie - for field in dict_with_new_values: + for field, new_value in dict_with_new_values.items(): try: current_value = cookie[field] except KeyError: current_value = None - new_value = dict_with_new_values[field] if new_value != current_value: print( @@ -78,6 +78,7 @@ def update_and_save_cookie_to_disk_if_values_changed( dict_with_new_values: dict[str, str], fields: list[str] | None = None, file_name_with_personal_info: str | None = None, + *, verbose: bool = False, ) -> dict[str, str]: if fields is None: diff --git a/src/sack_of_gems.py b/src/sack_of_gems.py index 8c356b1..861c032 100644 --- a/src/sack_of_gems.py +++ b/src/sack_of_gems.py @@ -19,6 +19,7 @@ def get_num_gems_per_sack_of_gems() -> int: def download_sack_of_gems_price( sack_of_gems_listing_file_name: str | None = None, + *, verbose: bool = True, ) -> float: if sack_of_gems_listing_file_name is None: @@ -38,7 +39,7 @@ def download_sack_of_gems_price( bid_price, ask_price, bid_volume, ask_volume = download_market_order_data( listing_hash, item_nameid, - verbose, + verbose=verbose, ) listing_details[listing_hash]["bid"] = bid_price listing_details[listing_hash]["ask"] = ask_price @@ -55,6 +56,7 @@ def download_sack_of_gems_price( def load_sack_of_gems_price( + *, retrieve_gem_price_from_scratch: bool = False, verbose: bool = True, sack_of_gems_listing_file_name: str | None = None, @@ -87,12 +89,13 @@ def load_sack_of_gems_price( def get_gem_price( enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, + *, retrieve_gem_price_from_scratch: bool = False, verbose: bool = True, ) -> float: if enforced_sack_of_gems_price is None: sack_of_gems_price = load_sack_of_gems_price( - retrieve_gem_price_from_scratch, + retrieve_gem_price_from_scratch=retrieve_gem_price_from_scratch, verbose=verbose, ) else: @@ -115,6 +118,7 @@ def get_gem_price( def print_gem_price_reminder( enforced_sack_of_gems_price: float | None = None, minimum_allowed_sack_of_gems_price: float | None = None, + *, retrieve_gem_price_from_scratch: bool | None = None, ) -> None: if retrieve_gem_price_from_scratch is None: diff --git a/src/tag_utils.py b/src/tag_utils.py new file mode 100644 index 0000000..9a5224b --- /dev/null +++ b/src/tag_utils.py @@ -0,0 +1,15 @@ +def get_tag_drop_rate_str(rarity: str | None = None) -> str: + if rarity is None: + rarity = "common" + + if rarity == "extraordinary": + tag_drop_rate_no = 3 + elif rarity == "rare": + tag_drop_rate_no = 2 + elif rarity == "uncommon": + tag_drop_rate_no = 1 + else: + # Rarity: Common + tag_drop_rate_no = 0 + + return f"tag_droprate_{tag_drop_rate_no}" diff --git a/src/utils.py b/src/utils.py index 5c3920f..3bd775a 100644 --- a/src/utils.py +++ b/src/utils.py @@ -1,5 +1,7 @@ from pathlib import Path +from src.tag_utils import get_tag_drop_rate_str + TIMEOUT_IN_SECONDS = 5 @@ -17,7 +19,7 @@ def get_data_folder() -> str: return data_folder -def get_badge_creation_file_name(from_javascript: bool = False) -> str: +def get_badge_creation_file_name(*, from_javascript: bool = False) -> str: badge_creation_file_name = get_data_folder() + "booster_game_creator" if from_javascript: @@ -36,8 +38,6 @@ def get_listing_output_file_name_suffix( tag_drop_rate_str: str | None = None, rarity: str | None = None, ) -> str: - from src.market_search import get_tag_drop_rate_str - if tag_drop_rate_str is None: tag_drop_rate_str = get_tag_drop_rate_str(rarity=rarity) @@ -185,7 +185,7 @@ def get_category_name_for_emoticons() -> str: return "emoticons" -def get_bullet_point_for_display(use_numbered_bullet_points: bool = False) -> str: +def get_bullet_point_for_display(*, use_numbered_bullet_points: bool = False) -> str: # Return a string, which consists of a bullet point followed by three spaces, to display lists in Markdown format. # # NB: if the list of bullet points is long, Numbered bullet points improve readability on GitHub Gist. diff --git a/tests.py b/tests.py index f73f680..bc33f60 100644 --- a/tests.py +++ b/tests.py @@ -17,7 +17,8 @@ class TestMarketListingMethods(unittest.TestCase): - def test_get_listing_details_batch(self) -> None: + @staticmethod + def test_get_listing_details_batch() -> None: listing_hashes = [ "407420-Gabe Newell Simulator Booster Pack", "443380-Tokyo Babel Booster Pack", @@ -31,46 +32,54 @@ def test_get_listing_details_batch(self) -> None: assert len(all_listing_details) == len(listing_hashes) - def test_main(self) -> None: + @staticmethod + def test_main() -> None: assert market_listing.main() is True class TestParsingUtilsMethods(unittest.TestCase): - def test_main(self) -> None: + @staticmethod + def test_main() -> None: assert parsing_utils.main() is True class TestCreationTimeUtilsMethods(unittest.TestCase): - def test_main(self) -> None: + @staticmethod + def test_main() -> None: assert creation_time_utils.main() is True class TestSackOfGemsMethods(unittest.TestCase): - def test_download_sack_of_gems_price(self) -> None: + @staticmethod + def test_download_sack_of_gems_price() -> None: sack_of_gems_price = sack_of_gems.download_sack_of_gems_price() assert sack_of_gems_price > 0 - def test_load_sack_of_gems_price(self) -> None: + @staticmethod + def test_load_sack_of_gems_price() -> None: sack_of_gems_price = sack_of_gems.load_sack_of_gems_price() assert sack_of_gems_price > 0 class TestMarketSearchMethods(unittest.TestCase): - def test_download_all_listings(self) -> None: + @staticmethod + def test_download_all_listings() -> None: assert market_search.download_all_listings() is True class TestMarketUtilsMethods(unittest.TestCase): - def test_load_aggregated_badge_data(self) -> None: + @staticmethod + def test_load_aggregated_badge_data() -> None: aggregated_badge_data = market_utils.load_aggregated_badge_data() assert aggregated_badge_data class TestMarketArbitrageMethods(unittest.TestCase): - def test_apply_workflow(self) -> None: + @staticmethod + def test_apply_workflow() -> None: try: flag = market_arbitrage.apply_workflow( retrieve_listings_from_scratch=False, @@ -84,7 +93,8 @@ def test_apply_workflow(self) -> None: class TestMarketOrderMethods(unittest.TestCase): - def test_main(self) -> None: + @staticmethod + def test_main() -> None: try: flag = market_order.main() except KeyError: @@ -95,22 +105,26 @@ def test_main(self) -> None: class TestUtilsMethods(unittest.TestCase): - def test_main(self) -> None: + @staticmethod + def test_main() -> None: assert utils.main() is True class TestTransactionFeeMethods(unittest.TestCase): - def test_main(self) -> None: + @staticmethod + def test_main() -> None: assert transaction_fee.main() is True class TestBatchCreatePacksMethods(unittest.TestCase): - def test_main(self) -> None: + @staticmethod + def test_main() -> None: assert batch_create_packs.main(is_a_simulation=True) is True class TestDropRateEstimatesMethods(unittest.TestCase): - def test_main(self) -> None: + @staticmethod + def test_main() -> None: assert drop_rate_estimates.main() is True