zman27/ai_giga_tcg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

part of pricing idk i dont remember

Browse Source
This commit is contained in:
zman
2025-06-09 08:28:14 -04:00
parent 7bc64115f2
commit 77d6fd6e29
4 changed files with 130 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
app/services/external_api/tcgplayer/tcgplayer_inventory_service.py
View File

@ -72,58 +72,3 @@ class TCGPlayerInventoryService(BaseTCGPlayerService):
with transaction(db):
# Bulk insert new data
db.bulk_insert_mappings(TCGPlayerInventory, inventory_data)
async def refresh_unmanaged_tcgplayer_inventory_table(self, db: Session):
"""
Refresh the TCGPlayer unmanaged inventory table
unmanaged inventory is any inventory that cannot be mapped to a card with a marketplace listing
"""
with transaction(db):
# Get active marketplace listings with their physical items in a single query
listed_cards = (
db.query(MarketplaceListing)
.join(MarketplaceListing.inventory_item)
.join(InventoryItem.physical_item)
.filter(
func.lower(Marketplace.name) == func.lower("tcgplayer"),
MarketplaceListing.delisting_date == None,
MarketplaceListing.deleted_at == None,
MarketplaceListing.listing_date != None
)
.all()
)
# Get current inventory and create lookup dict
current_inventory = db.query(TCGPlayerInventory).all()
# Create a set of SKUs that have active listings
listed_skus = {
card.inventory_item.physical_item.tcgplayer_sku_id
for card in listed_cards
}
unmanaged_inventory = []
for inventory in current_inventory:
# Only include SKUs that have no active listings
if inventory.tcgplayer_sku_id not in listed_skus:
unmanaged_inventory.append({
"tcgplayer_inventory_id": inventory.id,
"tcgplayer_sku_id": inventory.tcgplayer_sku_id,
"product_line": inventory.product_line,
"set_name": inventory.set_name,
"product_name": inventory.product_name,
"title": inventory.title,
"number": inventory.number,
"rarity": inventory.rarity,
"condition": inventory.condition,
"tcg_market_price": inventory.tcg_market_price,
"tcg_direct_low": inventory.tcg_direct_low,
"tcg_low_price_with_shipping": inventory.tcg_low_price_with_shipping,
"tcg_low_price": inventory.tcg_low_price,
"total_quantity": inventory.total_quantity,
"add_to_quantity": inventory.add_to_quantity,
"tcg_marketplace_price": inventory.tcg_marketplace_price,
"photo_url": inventory.photo_url
})
db.bulk_insert_mappings(UnmanagedTCGPlayerInventory, unmanaged_inventory)

5
app/services/inventory_service.py
View File

@ -375,7 +375,7 @@ class MarketplaceListingService(BaseService[MarketplaceListing]):
async def create_marketplace_listing(self, db: Session, inventory_item: InventoryItem, marketplace: Marketplace) -> MarketplaceListing:
try:
with db_transaction(db):
recommended_price = await self.pricing_service.set_price(db, inventory_item)
recommended_price = await self.pricing_service.set_price_for_inventory_item(db, inventory_item)
logger.info(f"recommended_price: {recommended_price.price}")
marketplace_listing = MarketplaceListing(
inventory_item=inventory_item,
@ -392,9 +392,10 @@ class MarketplaceListingService(BaseService[MarketplaceListing]):
raise e
async def update_marketplace_listing_price(self, db: Session, marketplace_listing: MarketplaceListing) -> MarketplaceListing:
# idk if this was ever even finished so prob doesnt work idk
try:
with db_transaction(db):
marketplace_listing.listed_price = self.pricing_service.set_price(marketplace_listing.inventory_item)
marketplace_listing.listed_price = self.pricing_service.set_price_for_inventory_item(db, marketplace_listing.inventory_item)
db.flush()
return marketplace_listing

166
app/services/pricing_service.py
View File

@ -1,8 +1,11 @@
import logging
from dataclasses import dataclass
from typing import Optional
from sqlalchemy.orm import Session
from app.services.base_service import BaseService
from app.models.inventory_management import InventoryItem, MarketplaceListing
from app.models.inventory_management import InventoryItem, MarketplaceListing, PhysicalItem
from app.models.tcgplayer_inventory import TCGPlayerInventory
from app.models.tcgplayer_products import TCGPlayerPriceHistory, TCGPlayerProduct, MTGJSONSKU
from app.models.pricing import PricingEvent
from app.db.database import transaction
from decimal import Decimal
@ -11,24 +14,109 @@ from datetime import datetime
logger = logging.getLogger(__name__)
@dataclass
class PriceData:
cost_basis: Optional[Decimal]
market_price: Optional[Decimal]
tcg_low: Optional[Decimal]
tcg_mid: Optional[Decimal]
direct_low: Optional[Decimal]
listed_price: Optional[Decimal]
quantity: int
lowest_price_for_qty: Optional[Decimal]
velocity: Optional[Decimal]
age_of_inventory: Optional[int]
class PricingService(BaseService):
def __init__(self):
super().__init__(None)
async def get_unmanaged_inventory(self, db: Session):
#select * from tcgplayer_inventory ti where ti.tcgplayer_sku_id not in (select pi2.tcgplayer_sku_id from marketplace_listings ml
# join inventory_items ii on ml.inventory_item_id = ii.id
# join physical_items pi2 on ii.physical_item_id = pi2.id
# where ml.delisting_date is null and ml.deleted_at is null and ii.deleted_at is null and pi2.deleted_at is null);
unmanaged_inventory = db.query(TCGPlayerInventory).filter(
TCGPlayerInventory.tcgplayer_sku_id.notin_(
db.query(MarketplaceListing.inventory_item.physical_item.tcgplayer_sku_id).join(
InventoryItem, MarketplaceListing.inventory_item_id == InventoryItem.id
).join(
PhysicalItem, InventoryItem.physical_item_id == PhysicalItem.id
).filter(
MarketplaceListing.delisting_date.is_(None),
MarketplaceListing.deleted_at.is_(None),
InventoryItem.deleted_at.is_(None),
PhysicalItem.deleted_at.is_(None)
)
)
).all()
return unmanaged_inventory
async def set_price(self, db: Session, inventory_item: InventoryItem) -> float:
async def update_prices_for_unmanaged_inventory(self, db: Session):
unmanaged_inventory = await self.get_unmanaged_inventory(db)
for inventory in unmanaged_inventory:
inventory.tcg_marketplace_price = await self.set_price_for_unmanaged_inventory(db, inventory)
async def set_price_for_unmanaged_inventory(self, db: Session, inventory: TCGPlayerInventory):
# available columns
# id, tcgplayer_sku_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, created_at, updated_at
# get mtgjson sku
mtgjson_sku = db.query(MTGJSONSKU).filter(
MTGJSONSKU.tcgplayer_sku_id == inventory.tcgplayer_sku_id
).first()
if mtgjson_sku:
tcgplayer_product = mtgjson_sku.product
price_data = PriceData(
cost_basis=None,
market_price=Decimal(str(tcgplayer_product.most_recent_tcgplayer_price.market_price)) if tcgplayer_product.most_recent_tcgplayer_price else None,
tcg_low=Decimal(str(tcgplayer_product.most_recent_tcgplayer_price.low_price)) if tcgplayer_product.most_recent_tcgplayer_price else None,
tcg_mid=Decimal(str(tcgplayer_product.most_recent_tcgplayer_price.mid_price)) if tcgplayer_product.most_recent_tcgplayer_price else None,
direct_low=Decimal(str(tcgplayer_product.most_recent_tcgplayer_price.direct_low_price)) if tcgplayer_product.most_recent_tcgplayer_price else None,
listed_price=inventory.tcg_marketplace_price,
quantity=inventory.total_quantity,
lowest_price_for_qty=None,
velocity=None,
age_of_inventory=None
)
return await self.set_price(db, price_data)
else:
return None
async def set_price_for_inventory_item(self, db: Session, inventory_item: InventoryItem):
price_data = PriceData(
cost_basis=Decimal(str(inventory_item.cost_basis)),
market_price=Decimal(str(inventory_item.physical_item.sku.product.most_recent_tcgplayer_price.market_price)),
tcg_low=Decimal(str(inventory_item.physical_item.sku.product.most_recent_tcgplayer_price.low_price)),
tcg_mid=Decimal(str(inventory_item.physical_item.sku.product.most_recent_tcgplayer_price.mid_price)),
direct_low=Decimal(str(inventory_item.physical_item.sku.product.most_recent_tcgplayer_price.direct_low_price)),
listed_price=Decimal(str(inventory_item.marketplace_listing.listed_price)),
quantity=db.query(TCGPlayerInventory).filter(
TCGPlayerInventory.tcgplayer_sku_id == inventory_item.physical_item.tcgplayer_sku_id
).first().total_quantity if db.query(TCGPlayerInventory).filter(
TCGPlayerInventory.tcgplayer_sku_id == inventory_item.physical_item.tcgplayer_sku_id
).first() else 0,
lowest_price_for_qty=None,
velocity=None,
age_of_inventory=None
)
return await self.set_price(db, price_data, inventory_item)
async def set_price(self, db: Session, price_data: PriceData, inventory_item: InventoryItem=None):
"""
TODO This sets listed_price per inventory_item but listed_price can only be applied to a product
however, this may be desired on other marketplaces
when generating pricing file for tcgplayer, give the option to set min, max, avg price for product?
"""
# Fetch base pricing data
cost_basis = Decimal(str(inventory_item.cost_basis))
market_price = Decimal(str(inventory_item.physical_item.sku.product.most_recent_tcgplayer_price.market_price))
tcg_low = Decimal(str(inventory_item.physical_item.sku.product.most_recent_tcgplayer_price.low_price))
tcg_mid = Decimal(str(inventory_item.physical_item.sku.product.most_recent_tcgplayer_price.mid_price))
listed_price = Decimal(str(inventory_item.marketplace_listing.listed_price)) if inventory_item.marketplace_listing else None
cost_basis = price_data.cost_basis
market_price = price_data.market_price
tcg_low = price_data.tcg_low
tcg_mid = price_data.tcg_mid
listed_price = price_data.listed_price
logger.info(f"listed_price: {listed_price}")
logger.info(f"market_price: {market_price}")
@ -68,10 +156,7 @@ class PricingService(BaseService):
card_cost_margin_multiplier = Decimal('0.05')
# Fetch current total quantity in stock for SKU
quantity_record = db.query(TCGPlayerInventory).filter(
TCGPlayerInventory.tcgplayer_sku_id == inventory_item.physical_item.tcgplayer_sku_id
).first()
quantity_in_stock = quantity_record.total_quantity if quantity_record else 0
quantity_in_stock = price_data.quantity
# Determine quantity multiplier based on stock levels
if quantity_in_stock < 4:
@ -91,6 +176,8 @@ class PricingService(BaseService):
# limit shipping cost offset to 10% of market price
shipping_cost_offset = min(shipping_cost / average_cards_per_order, market_price * Decimal('0.1'))
if cost_basis is None:
cost_basis = tcg_low * Decimal('0.65')
# Calculate base price considering cost, shipping, fees, and margin targets
base_price = (cost_basis + shipping_cost_offset) / (
(Decimal('1.0') - marketplace_fee_percentage) - adjusted_target_margin
@ -111,10 +198,10 @@ class PricingService(BaseService):
adjusted_price = tcg_mid
price_used = "tcg mid"
price_reason = "adjusted price below tcg low"
elif adjusted_price > tcg_low and adjusted_price < (market_price * Decimal('0.8')) and adjusted_price < (tcg_mid * Decimal('0.8')):
elif adjusted_price > tcg_low and adjusted_price < (tcg_mid * Decimal('0.8')):
adjusted_price = tcg_mid
price_used = "tcg mid"
price_reason = f"adjusted price below 80% of market price and tcg mid"
price_reason = f"adjusted price below 80% of tcg mid"
else:
price_used = "adjusted price"
price_reason = "valid price assigned based on margin targets"
@ -142,41 +229,22 @@ class PricingService(BaseService):
price_reason = "adjusted price below price drop threshold"
# Record pricing event in database transaction
with transaction(db):
pricing_event = PricingEvent(
inventory_item_id=inventory_item.id,
price=float(adjusted_price),
price_used=price_used,
price_reason=price_reason,
free_shipping_adjustment=free_shipping_adjustment
)
db.add(pricing_event)
# delete previous pricing events for inventory item
if inventory_item.marketplace_listing and inventory_item.marketplace_listing.listed_price:
inventory_item.marketplace_listing.listed_price.deleted_at = datetime.now()
return pricing_event
def set_price_for_unmanaged_inventory(self, db: Session, tcgplayer_sku_id: int, quantity: int) -> float:
pass
def update_price_for_product(self, db: Session, tcgplayer_sku_id: int) -> list[PricingEvent]:
# get inventory items for sku
updated_prices = []
inventory_items = db.query(InventoryItem).filter(
InventoryItem.physical_item.tcgplayer_sku_id == tcgplayer_sku_id
).all()
for inventory_item in inventory_items:
pricing_event = self.set_price(db, inventory_item)
updated_prices.append(pricing_event)
return updated_prices
def set_price_for_product(self, db: Session, tcgplayer_sku_id: int) -> float:
# update price for all inventory items for sku
prices = self.update_price_for_product(db, tcgplayer_sku_id)
sum_prices = sum(price.price for price in prices)
average_price = sum_prices / len(prices)
return average_price
if inventory_item:
with transaction(db):
pricing_event = PricingEvent(
inventory_item_id=inventory_item.id,
price=float(adjusted_price),
price_used=price_used,
price_reason=price_reason,
free_shipping_adjustment=free_shipping_adjustment
)
db.add(pricing_event)
# delete previous pricing events for inventory item
if inventory_item.marketplace_listing and inventory_item.marketplace_listing.listed_price:
inventory_item.marketplace_listing.listed_price.deleted_at = datetime.now()
return pricing_event
else:
return adjusted_price
# BAD BAD BAD FIX PLS TODO

8
app/services/scheduler/scheduler_service.py
View File

@ -4,6 +4,7 @@ from app.services.base_service import BaseService
from sqlalchemy import text
import logging
from app.models.tcgplayer_inventory import UnmanagedTCGPlayerInventory, TCGPlayerInventory
from datetime import datetime
logger = logging.getLogger(__name__)
class SchedulerService(BaseService):
@ -67,7 +68,6 @@ class SchedulerService(BaseService):
db.flush()
await tcgplayer_inventory_service.refresh_tcgplayer_inventory_table(db)
db.flush()
await tcgplayer_inventory_service.refresh_unmanaged_tcgplayer_inventory_table(db)
async def start_scheduled_tasks(self, db):
"""Start all scheduled tasks"""
@ -93,6 +93,12 @@ class SchedulerService(BaseService):
db=db
)
# Run initial inventory refresh on startup if inventory update was not run today
# get last inventory update date
last_inventory_update = db.query(TCGPlayerInventory).order_by(TCGPlayerInventory.created_at.desc()).first()
if last_inventory_update is None or last_inventory_update.created_at.date() != datetime.now().date():
await self.refresh_tcgplayer_inventory_table(db)
self.scheduler.start()
logger.info("All scheduled tasks started")