Amazon 상품 가격 예측 봇 구축기(1) - 전처리

from typing import Optional from transformers import AutoTokenizer import re BASE_MODEL = "meta-llama/Meta-Llama-3.1-8B" MIN_TOKENS = 150 MAX_TOKENS = 160 MIN_CHARS = 300 CEILING_CHARS = MAX_TOKENS * 7 class Item: """ An Item is a cleaned, curated datapoint of a Product with a Price """ tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL, trust_remote_code=True) PREFIX = "Price is $" QUESTION = "How much does this cost to the nearest dollar?" REMOVALS = [ '"Batteries Included?": "No"', '"Batteries Included?": "Yes"', '"Batteries Required?": "No"', '"Batteries Required?": "Yes"', "By Manufacturer", "Item", "Date First", "Package", ":", "Number of", "Best Sellers", "Number", "Product " ] title: str price: float category: str token_count: int = 0 details: Optional[str] prompt: Optional[str] = None include = False def __init__(self, data, price): self.title = data['title'] self.price = price self.parse(data) def scrub_details(self): """ Clean up the details string by removing common text that doesn't add value """ details = self.details for remove in self.REMOVALS: details = details.replace(remove, "") return details def scrub(self, stuff): """ Clean up the provided text by removing unnecessary characters and whitespace Also remove words that are 7+ chars and contain numbers, as these are likely irrelevant product numbers ex: Clothes Dryer Drum Slide, General Electric, Hotpoint, WE1M333, WE1M504 -> Clothes Dryer Drum Slide, General Electric, Hotpoint, """ stuff = re.sub(r'[:\[\]"{}【】\s]+', ' ', stuff).strip() stuff = stuff.replace(" ,", ",").replace(",,,",",").replace(",,",",") words = stuff.split(' ') select = [word for word in words if len(word)<7 or not any(char.isdigit() for char in word)] return " ".join(select) def parse(self, data): """ Parse this datapoint and if it fits within the allowed Token range, then set include to True """ contents = '\n'.join(data['description']) if contents: contents += '\n' features = '\n'.join(data['features']) if features: contents += features + '\n' self.details = data['details'] if self.details: contents += self.scrub_details() + '\n' if len(contents) > MIN_CHARS: contents = contents[:CEILING_CHARS] text = f"{self.scrub(self.title)}\n{self.scrub(contents)}" tokens = self.tokenizer.encode(text, add_special_tokens=False) if len(tokens) > MIN_TOKENS: tokens = tokens[:MAX_TOKENS] text = self.tokenizer.decode(tokens) self.make_prompt(text) self.include = True def make_prompt(self, text): """ Set the prompt instance variable to be a prompt appropriate for training """ self.prompt = f"{self.QUESTION}\n\n{text}\n\n" self.prompt += f"{self.PREFIX}{str(round(self.price))}.00" self.token_count = len(self.tokenizer.encode(self.prompt, add_special_tokens=False)) def test_prompt(self): """ Return a prompt suitable for testing, with the actual price removed """ return self.prompt.split(self.PREFIX)[0] + self.PREFIX

{'main_category': 'Tools & Home Improvement', 'title': 'Clothes Dryer Drum Slide, General Electric, Hotpoint, WE1M333, WE1M504', 'average_rating': 3.5, 'rating_number': 18, 'features': [], 'description': ['Brand new dryer drum slide, replaces General Electric, Hotpoint, RCA, WE1M333, WE1M504.'], 'price': 'None', 'images': {'hi_res': ['https://m.media-amazon.com/images/I/51TIpnkrEpL._AC_SL1500_.jpg'], 'large': ['https://m.media-amazon.com/images/I/21HQKcHPIkL._AC_.jpg'], 'thumb': ['https://m.media-amazon.com/images/I/21HQKcHPIkL._AC_US75_.jpg'], 'variant': ['MAIN']}, 'videos': {'title': [], 'url': [], 'user_id': []}, 'store': 'GE', 'categories': ['Appliances', 'Parts & Accessories'], 'details': '{"Manufacturer": "RPI", "Part Number": "WE1M333,", "Item Weight": "0.352 ounces", "Package Dimensions": "5.5 x 4.7 x 0.4 inches", "Item model number": "WE1M333,", "Is Discontinued By Manufacturer": "No", "Item Package Quantity": "1", "Batteries Included?": "No", "Batteries Required?": "No", "Best Sellers Rank": {"Tools & Home Improvement": 1315213, "Parts & Accessories": 181194}, "Date First Available": "February 25, 2014"}', 'parent_asin': 'B00IN9AGAE', 'bought_together': None, 'subtitle': None, 'author': None}

from typing import List, Optional from tqdm import tqdm from datasets import load_dataset from concurrent.futures import ProcessPoolExecutor from items import Item CHUNK_SIZE = 1000 MIN_PRICE = 0.5 MAX_PRICE = 999.49 class ItemLoader: def __init__(self, name): self.name = name self.dataset = None def from_datapoint(self, datapoint) -> Optional[Item]: """ Try to create an Item from this datapoint Return the Item if successful, or None if it shouldn't be included """ try: price_str = datapoint['price'] if price_str: price = float(price_str) if MIN_PRICE <= price <= MAX_PRICE: item = Item(datapoint, price) return item if item.include else None except ValueError: return None def from_chunk(self, chunk): """ Create a list of Items from this chunk of elements from the Dataset """ batch = [] for datapoint in chunk: result = self.from_datapoint(datapoint) if result: batch.append(result) return batch def chunk_generator(self): """ Iterate over the Dataset, yielding chunks of datapoints at a time """ size = len(self.dataset) for i in range(0, size, CHUNK_SIZE): yield self.dataset.select(range(i, min(i + CHUNK_SIZE, size))) def load_in_parallel(self, workers): """ Use concurrent.futures to farm out the work to process chunks of datapoints - This speeds up processing significantly, but will tie up your computer while it's doing so! """ results = [] chunk_count = (len(self.dataset) // CHUNK_SIZE) + 1 with ProcessPoolExecutor(max_workers=workers) as pool: for batch in tqdm(pool.map(self.from_chunk, self.chunk_generator()), total=chunk_count): results.extend(batch) for result in results: result.category = self.name return results def load(self, workers=8) -> List[Item]: """ Load in this dataset; the workers parameter specifies how many processes should work on loading and scrubbing the data """ print(f"Loading dataset {self.name}", flush=True) self.dataset = load_dataset("McAuley-Lab/Amazon-Reviews-2023", f"raw_meta_{self.name}", split="full", trust_remote_code=True) results = self.load_in_parallel(workers) return results

dataset_names = [ "Tools_and_Home_Improvement", "Cell_Phones_and_Accessories", "Health_and_Personal_Care", "Handmade_Products", "Appliances", "Musical_Instruments", ] items = [] for dataset_name in dataset_names: loader = ItemLoader(dataset_name) items.extend(loader.load()) print(f"A grand total of {len(items):,} items") >> A grand total of 944,154 items

# Plot the distribution of prices prices = [item.price for item in items] plt.figure(figsize=(15, 6)) plt.title(f"Prices: Avg {sum(prices)/len(prices):,.1f} and highest {max(prices):,}\n") plt.xlabel('Price ($)') plt.ylabel('Count') plt.hist(prices, rwidth=0.7, color="blueviolet", bins=range(0, 1000, 10)) plt.show()

category_counts = Counter() for item in items: category_counts[item.category] += 1 categories = category_counts.keys() counts = [category_counts[category] for category in categories] # Bar chart by category plt.figure(figsize=(15, 6)) plt.bar(categories, counts, color="goldenrod") plt.title('How many in each category') plt.xlabel('Categories') plt.ylabel('Count') plt.xticks(rotation=30, ha='right') # Add value labels on top of each bar for i, v in enumerate(counts): plt.text(i, v, f"{v:,}", ha='center', va='bottom') # Display the chart plt.show()

import pandas as pd slots = defaultdict(list) for item in items: slots[round(item.price)].append(item) len_slots = defaultdict(int) for price in slots.keys(): # From 1$ to 1000$ len_slots[price] = len(slots[price]) # Make DataFrame df = pd.DataFrame( data={ 'price': len_slots.keys(), 'num_items': len_slots.values() } ) # Plot the distribution of prices plt.figure(figsize=(15, 6)) plt.title(f"Number of Items: Avg {df.describe().loc['mean', 'num_items']:,.1f} and highest {df.describe().loc['max', 'num_items']:,}\n") plt.xlabel('Price ($)') plt.ylabel('Count') plt.xticks(ticks=[i for i in range(0, 1000, 100)]) plt.bar(x=df['price'], height=df['num_items'], width=0.7) plt.show()

def inverse_frequency_sampling(frequencies: dict) -> dict: values = np.array(list(frequencies.values())) inverse_weights = 1 / values normalized_weights = inverse_weights / np.sum(inverse_weights) weighted_categories = {category: weight for category, weight in zip(frequencies.keys(), normalized_weights)} return weighted_categories # Create a dataset called "sample" which tries to more evenly take from the range of prices # Set random seed for reproducibility weighted_categories = inverse_frequency_sampling(category_counts) np.random.seed(42) random.seed(42) sample = [] for i in range(1, 1000): slot = slots[i] if i >= 150: sample.extend(slot) elif len(slot) <= 1000: sample.extend(slot) else: weights = np.array([weighted_categories[item.category] for item in slot]) weights = weights / np.sum(weights) selected_indices = np.random.choice(len(slot), size=1000, replace=False, p=weights) selected = [slot[i] for i in selected_indices] sample.extend(selected) >> weighted_categories {'Tools_and_Home_Improvement': 0.0063880685753043434, 'Cell_Phones_and_Accessories': 0.014469315360038308, 'Health_and_Personal_Care': 0.7528361774639492, 'Handmade_Products': 0.05384522099950132, 'Appliances': 0.12074308788600839, 'Musical_Instruments': 0.051718129715198354} print(f"There are {len(sample):,} items in the sample") >> There are 198,092 items in the sample

random.seed(42) random.shuffle(sample) train = sample[30_000:50_000] validation = sample[200:300] test = sample[190_000:] print(f"Divided into a training set of {len(train):,} items, validation set of {len(validation):,} items, and test set of {len(test):,} items") >> Divided into a training set of 20,000 items, validation set of 100 items, and test set of 8,092 items

print(train[0].prompt) >> How much does this cost to the nearest dollar? Aluminum Manual Truck & Trailer Hand Crank Tarp Kit w/wo Tarp (Hand Crank ONLY (NO TARP)) All Truck Products EZ-Mount Dump Tarp Roller Kits are built for dump truck and dump trailer tarp applications. The extruded aluminum roller axle telescopes to the exact width of your truck. The tarp can attach to the roller with a tarp pocket or by preformed slot in the roller to bolt through the tarp's grommets. The kit includes crank handle, mounting brackets with bearings, tarp roller, Mesh Tarp, 20 ft pull-rope, 4 rope storing J-hooks, two rubber tarp straps with S-hooks, and mounting and tarp bolts. Dimensions 47 x 5.5 x 5. Price is $144.00

print(test[0].test_prompt()) >> How much does this cost to the nearest dollar? KOHLER Triton Centerset Lavatory Faucet, Polished Chrome (Handles Not Included) From the Manufacturer With practical design and solid brass construction, Triton faucets are an exceptional value. Competitively priced Triton faucets feature washerless ceramic valving, a durable Polished Chrome finish and vandal-resistant index buttons. The two-handle centerset faucet will hold up to years of daily use. Select from a choice of lever, cross, square or wristblade handle style options. (Handles Not Included) Two-handle centerset lavatory faucet for 4-inch centers KOHLER ceramic disc valves exceed industry longevity standards two times for a lifetime of durable performance Solid brass construction for durability and reliability Choose from lever, cross, square, or wristblade handle options For installation on 4-inch Price is $

train_prompts = [item.prompt for item in train] train_prices = [item.price for item in train] validation_prmopts = [item.prompt for item in validation] validation_prices = [item.price for item in validation] test_prompts = [item.test_prompt() for item in test] test_prices = [item.price for item in test] # Create a Dataset from the lists train_dataset = Dataset.from_dict({"text": train_prompts, "price": train_prices}) validation_dataset = Dataset.from_dict({"text": validation_prmopts, "price": validation_prices}) test_dataset = Dataset.from_dict({"text": test_prompts, "price": test_prices}) dataset = DatasetDict({ "train": train_dataset, "validation": validation_dataset, "test": test_dataset }) HF_USER = "jeffrey03" DATASET_NAME = f"{HF_USER}/amazon-item-price-data-lite-version" dataset.push_to_hub(DATASET_NAME, private=False)

with open('train.pkl', 'wb') as file: pickle.dump(train, file) with open('test.pkl', 'wb') as file: pickle.dump(test, file)