from sqlalchemy.orm import Session from app.db.models import File, CardTCGPlayer, Price, TCGPlayerInventory from app.services.util._dataframe import TCGPlayerPricingRow, DataframeUtil from app.services.file import FileService from app.services.tcgplayer import TCGPlayerService from uuid import uuid4 from app.db.utils import db_transaction from typing import List, Dict from decimal import Decimal, ROUND_HALF_UP import pandas as pd import numpy as np import logging logger = logging.getLogger(__name__) ACTIVE_PRICING_ALGORITHIM = 'tcgplayer_recommended_algo' # 'default_pricing_algo' or 'tcgplayer_recommended_algo' FREE_SHIPPING = True class PricingService: def __init__(self, db: Session, file_service: FileService, tcgplayer_service: TCGPlayerService): self.db = db self.file_service = file_service self.tcgplayer_service = tcgplayer_service self.df_util = DataframeUtil() # function for taking a tcgplayer pricing export with all set ids and loading it into the price table # can be run as needed or scheduled def get_pricing_export_content(self, file: File = None) -> bytes: if ACTIVE_PRICING_ALGORITHIM == 'default_pricing_algo' and FREE_SHIPPING == True: print("asdf") if file: file_content = self.file_service.get_file_content(file.id) else: file = self.tcgplayer_service.get_pricing_export_for_all_products() file_content = self.file_service.get_file_content(file.id) return file_content def load_pricing_csv_content_to_db(self, file_content: bytes): try: if not file_content: raise ValueError("No file content provided") price_types = { "tcg_market_price": "tcg_market_price", "tcg_direct_low": "tcg_direct_low", "tcg_low_price_with_shipping": "tcg_low_price_with_shipping", "tcg_low_price": "tcg_low_price", "tcg_marketplace_price": "listed_price" } required_columns = ["tcgplayer_id"] + list(price_types.keys()) df = self.df_util.csv_bytes_to_df(file_content) # Validate columns missing_columns = set(required_columns) - set(df.columns) if missing_columns: raise ValueError(f"Missing required columns: {missing_columns}") # Process in true batches for i in range(0, len(df), 1000): batch = df.iloc[i:i+1000] pricing_rows = [TCGPlayerPricingRow(row) for _, row in batch.iterrows()] # Query cards for this batch only tcgplayer_ids = [row.tcgplayer_id for row in pricing_rows] batch_cards = self.db.query(CardTCGPlayer).filter( CardTCGPlayer.tcgplayer_id.in_(tcgplayer_ids) ).all() existing_cards = {card.tcgplayer_id: card for card in batch_cards} new_prices = [] for row in pricing_rows: if row.tcgplayer_id not in existing_cards: continue card = existing_cards[row.tcgplayer_id] row_prices = [ Price( id=str(uuid4()), product_id=card.product_id, marketplace_id=None, type=price_type, # Added missing price_type price=getattr(row, col_name) ) for col_name, price_type in price_types.items() if getattr(row, col_name, None) is not None and getattr(row, col_name) > 0 ] new_prices.extend(row_prices) # Save each batch separately if new_prices: with db_transaction(self.db): self.db.bulk_save_objects(new_prices) except Exception as e: raise e # Consider adding logging here def cron_load_prices(self, file: File = None): file_content = self.get_pricing_export_content(file) self.tcgplayer_service.load_tcgplayer_cards(file_content) self.load_pricing_csv_content_to_db(file_content) def get_all_prices_for_products(self, product_ids: List[str]) -> Dict[str, Dict[str, float]]: all_prices = self.db.query(Price).filter( Price.product_id.in_(product_ids) ).all() price_lookup = {} for price in all_prices: if price.product_id not in price_lookup: price_lookup[price.product_id] = {} price_lookup[price.product_id][price.type] = price.price return price_lookup def apply_price_to_df_columns(self, row: pd.Series, price_lookup: Dict[str, Dict[str, float]]) -> pd.Series: product_prices = price_lookup.get(row['product_id'], {}) for price_type, price in product_prices.items(): row[price_type] = price return row def smooth_markup(self, price, markup_bands): """ Applies a smoothed markup based on the given price and markup bands. Uses numpy for smooth transitions. """ # Convert markup bands to lists for easy lookup markups = np.array(list(markup_bands.keys())) min_prices = np.array([x[0] for x in markup_bands.values()]) max_prices = np.array([x[1] for x in markup_bands.values()]) # Find the index of the price's range idx = np.where((min_prices <= price) & (max_prices >= price))[0] if len(idx) > 0: # If price is within a defined range, return the markup markup = markups[idx[0]] else: # If price is not directly within any range, check smooth transitions # Find the closest two bands for interpolation idx_lower = np.argmax(max_prices <= price) # Closest range below the price idx_upper = np.argmax(min_prices > price) # Closest range above the price if idx_lower != idx_upper: # Linear interpolation between the two neighboring markups price_diff = (price - max_prices[idx_lower]) / (min_prices[idx_upper] - max_prices[idx_lower]) markup = np.interp(price_diff, [0, 1], [markups[idx_lower], markups[idx_upper]]) # Apply the markup to the price return price * markup def tcgplayer_recommended_algo(self, row: pd.Series) -> pd.Series: # Convert input values to Decimal for precise arithmetic tcg_low = Decimal(str(row.get('tcg_low_price'))) if not pd.isna(row.get('tcg_low_price')) else None tcg_low_shipping = Decimal(str(row.get('tcg_low_price_with_shipping'))) if not pd.isna(row.get('tcg_low_price_with_shipping')) else None tcg_market_price = Decimal(str(row.get('tcg_market_price'))) if not pd.isna(row.get('tcg_market_price')) else None current_price = Decimal(str(row.get('tcg_marketplace_price'))) if not pd.isna(row.get('tcg_marketplace_price')) else None total_quantity = str(row.get('total_quantity')) if not pd.isna(row.get('total_quantity')) else "0" added_quantity = str(row.get('add_to_quantity')) if not pd.isna(row.get('add_to_quantity')) else "0" quantity = int(total_quantity) + int(added_quantity) if tcg_market_price is None: logger.warning(f"Missing pricing data for row: {row}") row['new_price'] = None return row TWO_PLACES = Decimal('0.01') # Original markup bands markup_bands = { Decimal('2.34'): (Decimal('0.01'), Decimal('0.50')), Decimal('1.36'): (Decimal('0.51'), Decimal('1.00')), Decimal('1.24'): (Decimal('1.01'), Decimal('3.00')), Decimal('1.15'): (Decimal('3.01'), Decimal('20.00')), Decimal('1.06'): (Decimal('20.01'), Decimal('35.00')), Decimal('1.05'): (Decimal('35.01'), Decimal('50.00')), Decimal('1.03'): (Decimal('50.01'), Decimal('100.00')), Decimal('1.02'): (Decimal('100.01'), Decimal('200.00')), Decimal('1.01'): (Decimal('200.01'), Decimal('1000.00')) } # Adjust markups if quantity is high if quantity > 3: adjusted_bands = {} increment = Decimal('0.20') for markup, price_range in zip(markup_bands.keys(), markup_bands.values()): new_markup = Decimal(str(markup)) + increment adjusted_bands[new_markup] = price_range increment -= Decimal('0.02') markup_bands = adjusted_bands #if FREE_SHIPPING: #if tcg_low_shipping and (tcg_low >= Decimal('5.00')): #tcg_compare_price = tcg_low_shipping #elif tcg_low_shipping and (tcg_low < Decimal('5.00')): #tcg_compare_price = max(tcg_low_shipping - Decimal('1.31'), tcg_low) #elif tcg_low: #tcg_compare_price = tcg_low #else: #logger.warning(f"No TCG low or shipping price available for row: {row}") #row['new_price'] = None #return row #else: #tcg_compare_price = tcg_low #if tcg_compare_price is None: #logger.warning(f"No TCG low price available for row: {row}") #row['new_price'] = None #return row tcg_compare_price = tcg_low # Apply the smoothed markup new_price = self.smooth_markup(tcg_compare_price, markup_bands) # Enforce minimum price if new_price < Decimal('0.35'): new_price = Decimal('0.25') # Avoid huge price drops if current_price is not None and Decimal(str(((current_price - new_price) / current_price))) > Decimal('0.5'): logger.warning(f"Price drop too large for row: {row}") new_price = current_price # Round to 2 decimal places new_price = new_price.quantize(TWO_PLACES, rounding=ROUND_HALF_UP) # Convert back to float for dataframe row['new_price'] = float(new_price) logger.debug(f""" card: {row['product_name']} TCGplayer Id: {row['tcgplayer_id']} Algorithm: {ACTIVE_PRICING_ALGORITHIM} TCG Low: {tcg_low} TCG Low Shipping: {tcg_low_shipping} TCG Market Price: {tcg_market_price} Current Price: {current_price} Total Quantity: {total_quantity} Added Quantity: {added_quantity} Quantity: {quantity} TCG Compare Price: {tcg_compare_price} New Price: {new_price} """) return row def default_pricing_algo(self, row: pd.Series) -> pd.Series: """Default pricing algorithm with complex pricing rules""" # Convert input values to Decimal for precise arithmetic tcg_low = Decimal(str(row.get('tcg_low_price'))) if not pd.isna(row.get('tcg_low_price')) else None tcg_low_shipping = Decimal(str(row.get('tcg_low_price_with_shipping'))) if not pd.isna(row.get('tcg_low_price_with_shipping')) else None tcg_market_price = Decimal(str(row.get('tcg_market_price'))) if not pd.isna(row.get('tcg_market_price')) else None total_quantity = str(row.get('total_quantity')) if not pd.isna(row.get('total_quantity')) else "0" added_quantity = str(row.get('add_to_quantity')) if not pd.isna(row.get('add_to_quantity')) else "0" quantity = int(total_quantity) + int(added_quantity) if tcg_market_price is None: logger.warning(f"Missing pricing data for row: {row}") row['new_price'] = None return row # Define precision for rounding TWO_PLACES = Decimal('0.01') # Apply pricing rules if tcg_market_price < Decimal('1') and tcg_market_price > Decimal('0.25'): new_price = tcg_market_price * Decimal('1.25') elif tcg_market_price < Decimal('0.25'): new_price = Decimal('0.25') elif tcg_market_price < Decimal('5'): new_price = tcg_market_price * Decimal('1.08') elif tcg_market_price < Decimal('10'): new_price = tcg_market_price * Decimal('1.06') elif tcg_market_price < Decimal('20'): new_price = tcg_market_price * Decimal('1.0125') elif tcg_market_price < Decimal('50'): new_price = tcg_market_price * Decimal('0.99') elif tcg_market_price < Decimal('100'): new_price = tcg_market_price * Decimal('0.98') else: new_price = tcg_market_price * Decimal('1.09') if new_price < Decimal('0.25'): new_price = Decimal('0.25') if quantity > 3: new_price = new_price * Decimal('1.1') # Ensure exactly 2 decimal places new_price = new_price.quantize(TWO_PLACES, rounding=ROUND_HALF_UP) # Convert back to float or string as needed for your dataframe row['new_price'] = float(new_price) return row def apply_pricing_algo(self, row: pd.Series, pricing_algo: callable = None) -> pd.Series: """Modified to handle the pricing algorithm as an instance method""" if pricing_algo: logger.debug(f"Using custom pricing algorithm: {pricing_algo.__name__}") return pricing_algo(row) elif ACTIVE_PRICING_ALGORITHIM == 'default_pricing_algo': logger.debug(f"Using default pricing algorithm: {self.default_pricing_algo.__name__}") pricing_algo = self.default_pricing_algo elif ACTIVE_PRICING_ALGORITHIM == 'tcgplayer_recommended_algo': logger.debug(f"Using TCGPlayer recommended algorithm: {self.tcgplayer_recommended_algo.__name__}") pricing_algo = self.tcgplayer_recommended_algo else: logger.debug(f"Using default pricing algorithm: {self.default_pricing_algo.__name__}") pricing_algo = self.default_pricing_algo return pricing_algo(row) def generate_tcgplayer_inventory_update_file_with_pricing(self, open_box_ids: List[str] = None) -> bytes: desired_columns = [ 'TCGplayer Id', 'Product Line', 'Set Name', 'Product Name', 'Title', 'Number', 'Rarity', 'Condition', 'TCG Market Price', 'TCG Direct Low', 'TCG Low Price With Shipping', 'TCG Low Price', 'Total Quantity', 'Add to Quantity', 'TCG Marketplace Price', 'Photo URL' ] if open_box_ids: # Get initial dataframe update_type = 'add' df = self.tcgplayer_service.open_box_cards_to_tcgplayer_inventory_df(open_box_ids) else: update_type = 'update' df = self.tcgplayer_service.get_inventory_df('live') # remove rows with total quantity of 0 df = df[df['total_quantity'] != 0] tcgplayer_ids = df['tcgplayer_id'].unique().tolist() # Make a single query to get all matching records product_id_mapping = { card.tcgplayer_id: card.product_id for card in self.db.query(CardTCGPlayer) .filter(CardTCGPlayer.tcgplayer_id.in_(tcgplayer_ids)) .all() } # Map tcgplayer_id to product_id, ensure strings, keep None if not found df['product_id'] = df['tcgplayer_id'].map(product_id_mapping).apply(lambda x: str(x) if pd.notnull(x) else None) # Log any tcgplayer_ids that didn't map to a product_id null_product_ids = df[df['product_id'].isnull()]['tcgplayer_id'].tolist() if null_product_ids: logger.warning(f"The following tcgplayer_ids could not be mapped to a product_id: {null_product_ids}") price_lookup = self.get_all_prices_for_products(df['product_id'].unique()) # Apply price columns df = df.apply(lambda row: self.apply_price_to_df_columns(row, price_lookup), axis=1) logger.debug(f"Applying pricing algorithm: {ACTIVE_PRICING_ALGORITHIM}") # set listed price df['listed_price'] = df['tcg_marketplace_price'].copy() # Apply pricing algorithm df = df.apply(self.apply_pricing_algo, axis=1) # if update type is update, remove rows where new_price == listed_price if update_type == 'update': df = df[df['new_price'] != df['listed_price']] # Set marketplace price df['TCG Marketplace Price'] = df['new_price'] df['Title'] = '' column_mapping = { 'tcgplayer_id': 'TCGplayer Id', 'product_line': 'Product Line', 'set_name': 'Set Name', 'product_name': 'Product Name', 'title': 'Title', 'number': 'Number', 'rarity': 'Rarity', 'condition': 'Condition', 'tcg_market_price': 'TCG Market Price', 'tcg_direct_low': 'TCG Direct Low', 'tcg_low_price_with_shipping': 'TCG Low Price With Shipping', 'tcg_low_price': 'TCG Low Price', 'total_quantity': 'Total Quantity', 'add_to_quantity': 'Add to Quantity', 'photo_url': 'Photo URL' } df = df.rename(columns=column_mapping) # Now do your column selection df = df[desired_columns] if update_type == 'update': with db_transaction(self.db): self.db.query(TCGPlayerInventory).delete() self.db.flush() # copy df to modify before inserting df_copy = df.copy() df_copy['id'] = df_copy.apply(lambda x: str(uuid4()), axis=1) # rename columns lowercase no space df_copy.columns = df_copy.columns.str.lower().str.replace(' ', '_') for index, row in df_copy.iterrows(): tcgplayer_inventory = TCGPlayerInventory(**row.to_dict()) self.db.add(tcgplayer_inventory) # remove any rows with no price #df = df[df['TCG Marketplace Price'] != 0] #df = df[df['TCG Marketplace Price'].notna()] # Convert to CSV bytes csv_bytes = self.df_util.df_to_csv_bytes(df) return csv_bytes