https://pastein.ru/t/1HN
Загрузка данных
import asyncio
import itertools
import logging
import random
import re
import sqlite3
from contextlib import closing
from pathlib import Path
from typing import Callable, Dict, Iterable, Iterator, List, Optional, TypeVar
from urllib.parse import quote, quote_plus
import time
from datetime import datetime
from curl_cffi.requests import AsyncSession
from funpayparsers.parsers.page_parsers import SubcategoryPageParser
from funpayparsers.parsers.page_parsers.offer_page_parser import OfferPageParser
import httpx
from nicegui import background_tasks, ui
from selectolax.parser import HTMLParser
from buff_cookie import BUFF_SESSION
try:
import emoji as emoji_lib
except ImportError:
emoji_lib = None
try:
from rapidfuzz import fuzz
except ImportError:
fuzz = None
# ==========================================
# 0. ЛОГИРОВАНИЕ
# ==========================================
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
datefmt='%H:%M:%S',
)
logger = logging.getLogger('arbitrage_scanner')
logger.setLevel(logging.DEBUG)
# ==========================================
# 0a. РОТАЦИЯ USER-AGENT И СТРУКТУРА ПОД ПРОКСИ
# ==========================================
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15',
]
def get_random_user_agent() -> str:
"""Случайный User-Agent для очередного запроса/сессии."""
return random.choice(USER_AGENTS)
PROXIES: List[str] = []
_proxy_cycle: Optional[Iterator[str]] = itertools.cycle(PROXIES) if PROXIES else None
def get_next_proxy() -> Optional[str]:
"""Возвращает следующий прокси по кругу (round-robin), либо None."""
if _proxy_cycle is None:
return None
return next(_proxy_cycle)
def httpx_proxy_kwarg(proxy_url: Optional[str]) -> dict:
if not proxy_url:
return {}
return {'proxy': proxy_url}
def curl_cffi_proxy_kwarg(proxy_url: Optional[str]) -> dict:
if not proxy_url:
return {}
return {'proxies': {'http': proxy_url, 'https': proxy_url}}
# ==========================================
# 0a1. ПОЛНЫЙ НАБОР БРАУЗЕРНЫХ ЗАГОЛОВКОВ ДЛЯ BUFF163
# ==========================================
BUFF_CHROME_MAJOR = '139'
BUFF_BROWSER_PROFILE = {
'user_agent': (
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
f' (KHTML, like Gecko) Chrome/{BUFF_CHROME_MAJOR}.0.0.0 Safari/537.36'
),
'sec_ch_ua': (
f'"Not;A=Brand"; v="99", "Google Chrome"; v="{BUFF_CHROME_MAJOR}", '
f'"Chromium"; v="{BUFF_CHROME_MAJOR}"'
),
'sec_ch_ua_platform': '"Windows"',
}
def build_buff_headers(game_code: str, session_cookie: str) -> Dict[str, str]:
return {
'User-Agent': BUFF_BROWSER_PROFILE['user_agent'],
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7',
'Referer': f'https://buff.163.com/market/{game_code}',
'X-Requested-With': 'XMLHttpRequest',
'sec-ch-ua': BUFF_BROWSER_PROFILE['sec_ch_ua'],
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': BUFF_BROWSER_PROFILE['sec_ch_ua_platform'],
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'same-origin',
'priority': 'u=1, i',
'Cookie': f'session={session_cookie}',
}
# ==========================================
# 0a2. ГЛОБАЛЬНЫЕ ЗАДЕРЖКИ
# ==========================================
BUFF_DELAY_RANGE = (5.5, 11.2)
FUNPAY_DELAY_RANGE = (2.1, 4.8)
STEAM_DELAY_RANGE = (5.0, 8.0)
STEAM_QUEUE_DELAY_RANGE = (10.0, 12.0)
STEAM_RATE_LIMIT_COOLDOWN_RANGE = (15.0, 25.0)
REQUEST_RETRY_DELAY_RANGE = (3.0, 7.5)
INTER_BATCH_DELAY_RANGE = (5.0, 15.0)
BUFF_PAGE_EXTRA_JITTER_RANGE = (0.3, 2.7)
# ==========================================
# 0b. ОБЩИЙ HTTP RETRY С ЭКСПОНЕНЦИАЛЬНЫМ BACKOFF
# ==========================================
T = TypeVar('T')
BACKOFF_MAX_RETRIES = 5
async def request_with_backoff(
request_fn,
*,
max_retries: int = BACKOFF_MAX_RETRIES,
base_delay: Optional[float] = None,
context: str = 'запрос',
):
if base_delay is None:
base_delay = random.uniform(*REQUEST_RETRY_DELAY_RANGE)
resp = await request_fn()
for attempt in range(max_retries):
status = getattr(resp, 'status_code', 200)
if status != 429 and not (500 <= status < 600):
return resp
delay = base_delay * (2 ** attempt) + random.uniform(0, 1.0)
logger.warning(
'%s: статус %s, попытка %s/%s, пауза %.1fc перед повтором',
context, status, attempt + 1, max_retries, delay,
)
await asyncio.sleep(delay)
resp = await request_fn()
logger.warning(
'%s: превышен лимит повторов (%s) после статуса %s, возвращаю как есть',
context, max_retries, getattr(resp, 'status_code', '?'),
)
return resp
def chunked(iterable: Iterable[T], size: int) -> Iterator[List[T]]:
it = iter(iterable)
while True:
batch = list(itertools.islice(it, size))
if not batch:
return
yield batch
# ==========================================
# 0c. ЛОКАЛЬНАЯ БД (SQLITE) КЭШ РЕЗУЛЬТАТОВ
# ==========================================
CACHE_DB_PATH = Path(__file__).with_name('scanner_cache.sqlite3')
CACHE_TTL_SECONDS = 20 * 60 # 20 минут
BATCH_SIZE = 25
def init_db(db_path: Path = CACHE_DB_PATH) -> None:
init_price_history_table(db_path)
with closing(sqlite3.connect(db_path)) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS funpay_offers (
offer_id TEXT PRIMARY KEY,
item_name TEXT NOT NULL,
price_rub REAL,
match_status TEXT NOT NULL,
updated_at REAL NOT NULL
)
''')
conn.execute('''
CREATE TABLE IF NOT EXISTS scan_results (
goods_id TEXT PRIMARY KEY,
item_name TEXT NOT NULL,
source TEXT NOT NULL,
price_rub REAL,
match_status TEXT NOT NULL,
updated_at REAL NOT NULL,
steam_volume INTEGER DEFAULT 0
)
''')
try:
conn.execute('ALTER TABLE scan_results ADD COLUMN steam_volume INTEGER DEFAULT 0')
except sqlite3.OperationalError:
pass
conn.execute('''
CREATE TABLE IF NOT EXISTS buff_prices (
goods_id TEXT PRIMARY KEY,
game TEXT,
item_name TEXT,
price_cny REAL,
price_rub REAL,
buy_max_cny REAL,
sell_num INTEGER,
updated_at REAL NOT NULL
)
''')
conn.execute('''
CREATE TABLE IF NOT EXISTS steam_prices (
goods_id TEXT PRIMARY KEY,
market_hash_name TEXT,
app_id INTEGER,
price_rub REAL,
volume_24h INTEGER DEFAULT 0,
updated_at REAL NOT NULL
)
''')
conn.execute('''
CREATE TABLE IF NOT EXISTS purchases (
id INTEGER PRIMARY KEY AUTOINCREMENT,
goods_id TEXT NOT NULL,
item_name TEXT NOT NULL,
game TEXT NOT NULL,
buy_price_rub REAL NOT NULL,
purchased_at REAL NOT NULL
)
''')
# Расширение схемы purchases (см. п.1 задачи): своя цена продажи +
# раздельные ссылки/метки времени по площадкам. ALTER TABLE ADD COLUMN
# оборачиваем в try/except — на уже существующей БД от предыдущих
# версий скрипта колонок ещё нет, а на свежей БД (CREATE TABLE только
# что их уже создал) ALTER упадёт с "duplicate column", что безопасно
# игнорируем — тот же паттерн, что уже используется для scan_results.
for column_sql in (
'ALTER TABLE purchases ADD COLUMN sell_price_rub REAL',
'ALTER TABLE purchases ADD COLUMN buff_url TEXT',
'ALTER TABLE purchases ADD COLUMN funpay_url TEXT',
'ALTER TABLE purchases ADD COLUMN buff_updated_at REAL',
'ALTER TABLE purchases ADD COLUMN funpay_updated_at REAL',
'ALTER TABLE purchases ADD COLUMN steam_updated_at REAL',
):
try:
conn.execute(column_sql)
except sqlite3.OperationalError:
pass
conn.commit()
def cache_get_offer(offer_id: int | str, db_path: Path = CACHE_DB_PATH) -> Optional[sqlite3.Row]:
with closing(sqlite3.connect(db_path)) as conn:
conn.row_factory = sqlite3.Row
cur = conn.execute('SELECT * FROM funpay_offers WHERE offer_id = ?', (str(offer_id),))
return cur.fetchone()
def cache_upsert_offer(
offer_id: int | str,
item_name: str,
price_rub: Optional[float],
match_status: str,
db_path: Path = CACHE_DB_PATH
) -> None:
with closing(sqlite3.connect(db_path)) as conn:
conn.execute('''
INSERT INTO funpay_offers (offer_id, item_name, price_rub, match_status, updated_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT (offer_id) DO UPDATE SET
item_name=excluded.item_name,
price_rub=excluded.price_rub,
match_status=excluded.match_status,
updated_at=excluded.updated_at
''', (str(offer_id), item_name, price_rub, match_status, time.time()))
conn.commit()
def cache_get_scan_result(goods_id: int | str, db_path: Path = CACHE_DB_PATH) -> Optional[sqlite3.Row]:
with closing(sqlite3.connect(db_path)) as conn:
conn.row_factory = sqlite3.Row
cur = conn.execute('SELECT * FROM scan_results WHERE goods_id = ?', (str(goods_id),))
return cur.fetchone()
def cache_upsert_scan_result(
goods_id: int | str,
item_name: str,
source: str,
price_rub: Optional[float],
match_status: str,
db_path: Path = CACHE_DB_PATH,
steam_volume: int = 0,
) -> None:
with closing(sqlite3.connect(db_path)) as conn:
conn.execute('''
INSERT INTO scan_results (goods_id, item_name, source, price_rub, match_status, updated_at, steam_volume)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (goods_id) DO UPDATE SET
item_name=excluded.item_name,
source=excluded.source,
price_rub=excluded.price_rub,
match_status=excluded.match_status,
updated_at=excluded.updated_at,
steam_volume=excluded.steam_volume
''', (str(goods_id), item_name, source, price_rub, match_status, time.time(), steam_volume))
conn.commit()
def is_cache_fresh(row: Optional[sqlite3.Row], ttl: float = CACHE_TTL_SECONDS) -> bool:
return row is not None and (time.time() - row['updated_at']) < ttl
def record_buff_snapshot(items: list, cny_rate: float, game: str, db_path: Path = CACHE_DB_PATH) -> int:
now = time.time()
history_rows = []
latest_rows = []
for item in items:
goods_id = item.get('id')
try:
price_cny = float(item.get('sell_min_price', 0.0) or 0.0)
except (TypeError, ValueError):
continue
if goods_id is None or price_cny <= 0:
continue
price_rub = price_cny * cny_rate
history_rows.append((str(goods_id), price_rub, now))
latest_rows.append((
str(goods_id), game,
item.get('market_hash_name') or item.get('name') or '',
price_cny, round(price_rub, 2),
float(item.get('buy_max_price', 0.0) or 0.0),
int(item.get('sell_num', 0) or 0), now,
))
if not history_rows:
return 0
with closing(sqlite3.connect(db_path)) as conn:
conn.executemany(
'INSERT INTO price_history (goods_id, price_rub, recorded_at) VALUES (?, ?, ?)',
history_rows,
)
conn.executemany('''
INSERT INTO buff_prices (goods_id, game, item_name, price_cny, price_rub, buy_max_cny, sell_num, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (goods_id) DO UPDATE SET
game=excluded.game,
item_name=excluded.item_name,
price_cny=excluded.price_cny,
price_rub=excluded.price_rub,
buy_max_cny=excluded.buy_max_cny,
sell_num=excluded.sell_num,
updated_at=excluded.updated_at
''', latest_rows)
conn.commit()
return len(history_rows)
def buff_price_get(goods_id: int | str, db_path: Path = CACHE_DB_PATH) -> Optional[sqlite3.Row]:
with closing(sqlite3.connect(db_path)) as conn:
conn.row_factory = sqlite3.Row
return conn.execute('SELECT * FROM buff_prices WHERE goods_id = ?', (str(goods_id),)).fetchone()
STEAM_PRICE_TTL_SECONDS = 60 * 60
def steam_cache_get(goods_id: int | str, db_path: Path = CACHE_DB_PATH) -> Optional[sqlite3.Row]:
with closing(sqlite3.connect(db_path)) as conn:
conn.row_factory = sqlite3.Row
return conn.execute('SELECT * FROM steam_prices WHERE goods_id = ?', (str(goods_id),)).fetchone()
def steam_cache_upsert(
goods_id: int | str, market_hash_name: str, app_id: int,
price_rub: float, volume_24h: int, db_path: Path = CACHE_DB_PATH
) -> None:
with closing(sqlite3.connect(db_path)) as conn:
conn.execute('''
INSERT INTO steam_prices (goods_id, market_hash_name, app_id, price_rub, volume_24h, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT (goods_id) DO UPDATE SET
market_hash_name=excluded.market_hash_name,
app_id=excluded.app_id,
price_rub=excluded.price_rub,
volume_24h=excluded.volume_24h,
updated_at=excluded.updated_at
''', (str(goods_id), market_hash_name, app_id, price_rub, volume_24h, time.time()))
conn.commit()
def steam_cache_is_fresh(goods_id: int | str) -> bool:
return is_cache_fresh(steam_cache_get(goods_id), ttl=STEAM_PRICE_TTL_SECONDS)
def purchase_add(
goods_id: int | str, item_name: str, game: str, buy_price_rub: float,
sell_price_rub: Optional[float] = None,
buff_url: Optional[str] = None,
funpay_url: Optional[str] = None,
db_path: Path = CACHE_DB_PATH,
) -> None:
now = time.time()
with closing(sqlite3.connect(db_path)) as conn:
conn.execute(
'''INSERT INTO purchases
(goods_id, item_name, game, buy_price_rub, sell_price_rub,
buff_url, funpay_url, purchased_at, buff_updated_at, funpay_updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''',
(
str(goods_id), item_name, game, float(buy_price_rub),
float(sell_price_rub) if sell_price_rub is not None else None,
buff_url, funpay_url, now,
now if buff_url else None,
now if funpay_url else None,
),
)
conn.commit()
def purchase_update_prices(
purchase_id: int,
buy_price_rub: Optional[float] = None,
sell_price_rub: Optional[float] = None,
db_path: Path = CACHE_DB_PATH,
) -> None:
"""
Точечное обновление цен покупки/продажи из редактируемых ячеек AG Grid
(п.3 задачи, onCellValueChanged) — трогает только переданные поля, не
задевая ссылки/метки времени, которые обновляются отдельно сканером
(см. sync_purchase_market_links).
"""
sets: List[str] = []
params: List[object] = []
if buy_price_rub is not None:
sets.append('buy_price_rub = ?')
params.append(float(buy_price_rub))
if sell_price_rub is not None:
sets.append('sell_price_rub = ?')
params.append(float(sell_price_rub))
if not sets:
return
params.append(int(purchase_id))
with closing(sqlite3.connect(db_path)) as conn:
conn.execute(f'UPDATE purchases SET {", ".join(sets)} WHERE id = ?', params)
conn.commit()
def sync_purchase_market_links(
goods_id: int | str,
buff_url: Optional[str] = None,
funpay_url: Optional[str] = None,
funpay_matched: bool = False,
steam_data_loaded: bool = False,
db_path: Path = CACHE_DB_PATH,
) -> None:
"""
Вызывается на КАЖДОЙ пересборке строки сканера (build_row) для товаров,
которые уже отслеживаются в "Избранном" (п.1, п.2 задачи) — обновляет
ссылку/метку времени по каждой площадке НЕЗАВИСИМО, а не всё сразу одним
updated_at: buff_url обновляется всегда (Buff163 отдаёт цену/лот в каждом
скане, если товар вообще есть в каталоге), funpay_url/funpay_updated_at —
только когда funpay_matched (в этом скане реально нашёлся оффер FunPay),
steam_updated_at — только когда steam_data_loaded (получен настоящий,
а не оценочный объём/цена Steam). Так в "Избранном" видно, если,
например, Steam обновился, а FunPay всё ещё показывает данные месячной
давности (потому что в последних сканах лот на FunPay не находился).
НЕ трогает buy_price_rub/sell_price_rub — это цены пользователя.
"""
now = time.time()
sets = ['buff_url = ?', 'buff_updated_at = ?']
params: List[object] = [buff_url, now]
if funpay_matched and funpay_url:
sets += ['funpay_url = ?', 'funpay_updated_at = ?']
params += [funpay_url, now]
if steam_data_loaded:
sets += ['steam_updated_at = ?']
params += [now]
params.append(str(goods_id))
with closing(sqlite3.connect(db_path)) as conn:
conn.execute(f'UPDATE purchases SET {", ".join(sets)} WHERE goods_id = ?', params)
conn.commit()
def purchase_delete(purchase_id: int, db_path: Path = CACHE_DB_PATH) -> None:
with closing(sqlite3.connect(db_path)) as conn:
conn.execute('DELETE FROM purchases WHERE id = ?', (int(purchase_id),))
conn.commit()
def purchases_list(db_path: Path = CACHE_DB_PATH) -> List[sqlite3.Row]:
with closing(sqlite3.connect(db_path)) as conn:
conn.row_factory = sqlite3.Row
return conn.execute('SELECT * FROM purchases ORDER BY purchased_at DESC').fetchall()
def purchased_goods_ids(db_path: Path = CACHE_DB_PATH) -> set:
with closing(sqlite3.connect(db_path)) as conn:
return {r[0] for r in conn.execute('SELECT DISTINCT goods_id FROM purchases')}
# ==========================================
# 0d. ИСТОРИЯ ЦЕН И ВОЛАТИЛЬНОСТЬ
# ==========================================
VOLATILITY_MIN_WINDOW_HOURS = 1.0
VOLATILITY_HIGH_WINDOW_HOURS = 3.0
VOLATILITY_LOW_WINDOW_HOURS = 24.0
VOLATILITY_LOOKBACK_HOURS = 24.0
VOLATILITY_HIGH_THRESHOLD_PCT = 5.0
VOLATILITY_LOW_THRESHOLD_PCT = 1.0
def init_price_history_table(db_path: Path = CACHE_DB_PATH) -> None:
with closing(sqlite3.connect(db_path)) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
goods_id TEXT NOT NULL,
price_rub REAL NOT NULL,
recorded_at REAL NOT NULL
)
''')
conn.execute('CREATE INDEX IF NOT EXISTS idx_price_history_goods_time ON price_history (goods_id, recorded_at)')
conn.commit()
def record_price_history(goods_id: int | str, price_rub: float, db_path: Path = CACHE_DB_PATH) -> None:
with closing(sqlite3.connect(db_path)) as conn:
conn.execute(
'INSERT INTO price_history (goods_id, price_rub, recorded_at) VALUES (?, ?, ?)',
(str(goods_id), price_rub, time.time()),
)
conn.commit()
def _get_price_history_since(goods_id: int | str, since_ts: float, db_path: Path = CACHE_DB_PATH) -> List[sqlite3.Row]:
with closing(sqlite3.connect(db_path)) as conn:
conn.row_factory = sqlite3.Row
cur = conn.execute(
'SELECT price_rub, recorded_at FROM price_history WHERE goods_id = ? AND recorded_at >= ? ORDER BY recorded_at ASC',
(str(goods_id), since_ts),
)
return cur.fetchall()
def compute_price_volatility_pct(prices: List[float]) -> float:
if len(prices) < 2:
return 0.0
mean = sum(prices) / len(prices)
if mean <= 0:
return 0.0
variance = sum((p - mean) ** 2 for p in prices) / len(prices)
std = variance ** 0.5
return (std / mean) * 100
def compute_dynamic_window_hours(goods_id: int | str, db_path: Path = CACHE_DB_PATH) -> float:
now = time.time()
rows = _get_price_history_since(goods_id, now - VOLATILITY_LOOKBACK_HOURS * 3600, db_path)
prices = [row['price_rub'] for row in rows if row['price_rub'] is not None]
if len(prices) < 2:
return VOLATILITY_HIGH_WINDOW_HOURS
volatility_pct = compute_price_volatility_pct(prices)
if volatility_pct >= VOLATILITY_HIGH_THRESHOLD_PCT:
window_hours = VOLATILITY_HIGH_WINDOW_HOURS
elif volatility_pct <= VOLATILITY_LOW_THRESHOLD_PCT:
window_hours = VOLATILITY_LOW_WINDOW_HOURS
else:
span = VOLATILITY_HIGH_THRESHOLD_PCT - VOLATILITY_LOW_THRESHOLD_PCT
t = (VOLATILITY_HIGH_THRESHOLD_PCT - volatility_pct) / span
window_hours = VOLATILITY_HIGH_WINDOW_HOURS + t * (VOLATILITY_LOW_WINDOW_HOURS - VOLATILITY_HIGH_WINDOW_HOURS)
return max(VOLATILITY_MIN_WINDOW_HOURS, window_hours)
def get_recent_price_history(goods_id: int | str, db_path: Path = CACHE_DB_PATH) -> tuple[List[sqlite3.Row], float]:
window_hours = compute_dynamic_window_hours(goods_id, db_path)
since_ts = time.time() - window_hours * 3600
rows = _get_price_history_since(goods_id, since_ts, db_path)
return rows, window_hours
STEAM_CACHE_TTL_SECONDS = 10 * 60
BUFF_GAME_CODES = {
'CS2': 'csgo',
'Dota 2': 'dota2',
'TF2': 'tf2',
}
STEAM_APP_IDS = {
'CS2': 730,
'Dota 2': 570,
'TF2': 440,
}
FUZZY_MATCH_THRESHOLD = 70
FUNPAY_CURRENCY_SYMBOLS = {
'₽': 'RUB',
'руб.': 'RUB',
'$': 'USD',
'€': 'EUR',
}
# ==========================================
# 1. МОДУЛЬ ЧИСТКИ И МАТЧИНГА НАЗВАНИЙ
# ==========================================
def extract_clean_name(offer) -> Optional[str]:
if not offer.title:
return None
raw = offer.title.split(',')[0].strip()
if emoji_lib is not None:
raw = emoji_lib.replace_emoji(raw, replace='')
raw = re.sub(r'[^\w\s()\-.]', '', raw, flags=re.UNICODE)
raw = re.sub(r'\s+', ' ', raw).strip()
return raw or None
def normalize_for_match(name: str) -> str:
return re.sub(r'\s+', '', name.lower().replace('™', '').replace('®', '')).strip()
_CJK_RE = re.compile(r'[\u4e00-\u9fff\u3400-\u4dbf\uf900-\ufaff]')
def has_cjk(text: str) -> bool:
return bool(_CJK_RE.search(text or ''))
DOTA_QUALITY_KEYWORDS = (
'standard', 'inscribed', 'autographed', 'frozen', 'infused',
'auspicious', 'heroic', 'genuine', 'cursed', 'base',
'corrupted', 'unusual', 'exalted', 'elder', 'legacy', 'ascendant',
)
NAME_MODIFIER_KEYWORDS = ('stattrak', 'souvenir', 'foil', 'golden', 'crimson') + DOTA_QUALITY_KEYWORDS
GEM_KEYWORDS = {
'ethereal', 'prismatic', 'kinetic', 'iridescent', 'emerald', 'aggregate',
'cracked', 'fervor', 'molten', 'frostbitten', 'diretide', 'foulfell',
'charmed', 'blighted', 'seasonal', 'radiant', 'spectral', 'spectator',
'iron', 'sacred', 'autograph', 'rune', 'gem', 'socket',
'гем', 'гема', 'гему', 'руна', 'призматический', 'призматическая',
'кинетический', 'кинетическая', 'эфирный', 'эфирная',
}
def modifiers_consistent(target_clean: str, candidate_clean: str) -> bool:
return all((kw in target_clean) == (kw in candidate_clean) for kw in NAME_MODIFIER_KEYWORDS)
FUNPAY_NON_SALE_TRADE_TYPES = {
'аренда', 'rent', 'обмен', 'exchange', 'trade', 'услуга', 'service', 'буст', 'boost',
}
def is_regular_sale(offer) -> bool:
trade_type = ((offer.other_data or {}).get('f-type2') or '').strip().lower()
if not trade_type:
return True
return trade_type not in FUNPAY_NON_SALE_TRADE_TYPES
FUNPAY_SCAM_KEYWORDS = (
'багнут', 'баганут', 'бажен', 'bugged', 'trail of burning doom',
'поменять эффект', 'любой эффект', 'любой из', 'аренда', 'rent', 'буст', 'сэт', 'сет',
)
def is_legit_offer(offer) -> bool:
title = offer.title or ''
description = getattr(offer, 'description', None) or ''
haystack = f'{title} {description}'.lower()
for kw in FUNPAY_SCAM_KEYWORDS:
if kw in haystack:
logger.debug('Лот %s отброшен: стоп-слово "%s" в "%s"', getattr(offer, 'id', '?'), kw, title)
return False
return True
def names_match(target_clean: str, candidate_clean: str, strict: bool = False) -> bool:
if target_clean == candidate_clean:
return True
if strict:
logger.debug('Нет совпадения (strict): "%s" != "%s"', target_clean, candidate_clean)
return False
if not modifiers_consistent(target_clean, candidate_clean):
logger.debug('Нет совпадения: разные модификаторы у "%s" и "%s"', target_clean, candidate_clean)
return False
if target_clean in candidate_clean:
return True
if fuzz is None:
return False
if len(candidate_clean) + 3 < len(target_clean):
return False
score = fuzz.token_sort_ratio(target_clean, candidate_clean)
if score < FUZZY_MATCH_THRESHOLD:
logger.debug('Нет совпадения: fuzzy-скор %.1f < %s для "%s" / "%s"', score, FUZZY_MATCH_THRESHOLD, target_clean, candidate_clean)
return False
return True
def extract_gem_hint(target_clean: str, candidate_clean: str) -> Optional[str]:
idx = candidate_clean.find(target_clean)
if idx == -1:
return None
tail = candidate_clean[idx + len(target_clean):].strip()
if not tail:
return None
if not any(kw in tail for kw in GEM_KEYWORDS):
return None
return ' '.join(tail.split()[:3]) or None
# ==========================================
# 2. МОДУЛЬ ПАРСИНГА И АВТОПОИСКА ССЫЛОК
# ==========================================
async def fetch_cny_rub_rate() -> float:
try:
async with httpx.AsyncClient(timeout=5.0) as client:
res = await client.get('https://open.er-api.com/v6/latest/CNY')
if res.status_code == 200:
data = res.json()
return float(data['rates'].get('RUB', 12.5))
except Exception as e:
logger.debug('Не удалось получить курс CNY -> RUB: %s', e)
return 12.5
async def fetch_fx_rates_to_rub() -> Dict[str, float]:
rates = {'RUB': 1.0, 'USD': 95.0, 'EUR': 105.0}
try:
async with httpx.AsyncClient(timeout=5.0) as client:
res = await client.get('https://open.er-api.com/v6/latest/RUB')
if res.status_code == 200:
rub_to = res.json().get('rates', {})
if rub_to.get('USD'):
rates['USD'] = 1 / rub_to['USD']
if rub_to.get('EUR'):
rates['EUR'] = 1 / rub_to['EUR']
except Exception as e:
logger.warning('Не удалось получить курсы валют для FunPay (USD/EUR->RUB): %s', e)
return rates
async def fetch_funpay_offers(url: str, session: AsyncSession) -> list:
try:
resp = await request_with_backoff(
lambda: session.get(url, timeout=10.0), context=f'FunPay catalog {url}'
)
if resp.status_code == 200:
parsed_page = SubcategoryPageParser(resp.text).parse()
return parsed_page.offers or []
except Exception as e:
logger.warning('Ошибка парсинга FunPay (%s): %s', url, e)
return []
def money_value_to_rub(price_obj, fx_rates: Dict[str, float]) -> float:
if hasattr(price_obj, 'value'):
raw_val = float(price_obj.value)
elif hasattr(price_obj, 'amount'):
raw_val = float(price_obj.amount)
else:
raw_val = float(price_obj)
currency_code = FUNPAY_CURRENCY_SYMBOLS.get(getattr(price_obj, 'character', None), 'RUB')
rate = fx_rates.get(currency_code, 1.0)
return raw_val * rate
def offer_price_in_rub(offer, fx_rates: Dict[str, float]) -> float:
return money_value_to_rub(offer.price, fx_rates)
FUNPAY_OFFER_PAGE_MAX_RETRIES = 2
REFINE_FUNPAY_PRICE_VIA_OFFER_PAGE = True
async def fetch_funpay_offer_details(
offer_id: int | str, fx_rates: Dict[str, float], session: AsyncSession
) -> Optional[tuple[float, dict]]:
url = f'https://funpay.com/lots/offer?id={offer_id}'
page = None
for attempt in range(FUNPAY_OFFER_PAGE_MAX_RETRIES + 1):
try:
resp = await request_with_backoff(
lambda: session.get(url, timeout=10.0), context=f'FunPay lot {offer_id}'
)
if resp.status_code != 200:
return None
page = OfferPageParser(resp.text).parse()
break
except Exception as e:
if attempt < FUNPAY_OFFER_PAGE_MAX_RETRIES:
delay = random.uniform(2.0, 4.0)
logger.warning('Обрыв при загрузке лота %s (попытка %s/%s): %s повтор через %.1fc', offer_id, attempt + 1, FUNPAY_OFFER_PAGE_MAX_RETRIES + 1, e, delay)
await asyncio.sleep(delay)
continue
logger.warning('Ошибка загрузки страницы лота %s: %s', offer_id, e)
return None
if page is None or not page.payment_options:
return None
prices_rub = [money_value_to_rub(opt.price, fx_rates) for opt in page.payment_options.values()]
if not prices_rub:
return None
return min(prices_rub), (page.fields or {})
def extract_target_quality(item_name: str) -> Optional[str]:
name_lower = item_name.lower()
for quality in DOTA_QUALITY_KEYWORDS:
if quality in name_lower:
return quality
return None
def get_min_funpay_offer(
item_name: str,
offers: list,
fx_rates: Dict[str, float],
strict: bool = False,
target_rarity: Optional[str] = None,
target_hero: Optional[str] = None,
target_quality: Optional[str] = None,
) -> Optional[tuple[float, int | str, Optional[str], int]]:
best: Optional[tuple[float, int | str, Optional[str]]] = None
match_count = 0
target = normalize_for_match(item_name)
# ВАЖНО: target_words нужно строить из "сырого" названия, ДО
# normalize_for_match — та убирает вообще все пробелы (не схлопывает, а
# удаляет), поэтому target.split() возвращал один слитный "псевдослов"
# (например 'infernalchieftain'), которого, разумеется, нет ни в одном
# заголовке лота как подстроки — предфильтр ниже отбраковывал из-за этого
# почти любой реальный лот. Раздельные слова нужно брать из исходного
# item_name (с пробелами), в нижнем регистре и без пунктуации.
target_words_source = re.sub(r'[^\w\s]', ' ', item_name.lower(), flags=re.UNICODE)
target_words = [w for w in target_words_source.split() if len(w) >= 3] or target_words_source.split()
for offer in offers:
title_lower = (offer.title or '').lower()
offer_id = getattr(offer, 'id', '?')
if target_words and not all(w in title_lower for w in target_words):
#missing_words = [w for w in target_words if w not in title_lower]
#logger.debug(
# 'Лот %s отброшен предфильтром по словам: заголовок "%s", не найдены слова: %s',
# offer_id, title_lower, missing_words,
#)
continue
if not is_regular_sale(offer):
logger.debug('Лот %s отброшен: не обычная продажа (f-type2)', offer_id)
continue
if not is_legit_offer(offer):
continue
candidate_name = extract_clean_name(offer)
if not candidate_name:
continue
candidate_clean = normalize_for_match(candidate_name)
is_exact = candidate_clean == target
other_data = offer.other_data or {}
if not is_exact:
if target_rarity:
offer_rarity = (other_data.get('f-rare') or '').strip().lower()
if offer_rarity and offer_rarity != target_rarity:
logger.debug('Лот %s отброшен: редкость "%s" != "%s"', offer_id, offer_rarity, target_rarity)
continue
if target_hero:
offer_hero = (other_data.get('f-hero') or '').strip().lower()
if offer_hero and offer_hero != target_hero:
logger.debug('Лот %s отброшен: герой "%s" != "%s"', offer_id, offer_hero, target_hero)
continue
if target_quality:
offer_quality = (other_data.get('f-quality') or '').strip().lower()
if offer_quality and offer_quality != target_quality:
logger.debug('Лот %s отброшен: качество "%s" != "%s"', offer_id, offer_quality, target_quality)
continue
if not is_exact and not names_match(target, candidate_clean, strict=strict):
continue
match_count += 1
price_rub = offer_price_in_rub(offer, fx_rates)
if best is None or price_rub < best[0]:
gem_hint = None if is_exact else extract_gem_hint(target, candidate_clean)
best = (price_rub, offer.id, gem_hint)
if best is None:
return None
return (*best, match_count)
GEM_PRICE_ESTIMATE_PCT = 0.15
async def discover_funpay_urls() -> Dict[str, str]:
targets = {
'CS2': {'keywords': ['Counter-Strike', 'CS2', 'CS:GO'], 'sections': ['Скины', 'Кейсы', 'Предметы']},
'Dota 2': {'keywords': ['Dota 2'], 'sections': ['Предметы']},
'TF2': {'keywords': ['Team Fortress 2', 'TF2'], 'sections': ['Предметы']}
}
found_urls = {}
try:
headers = {'User-Agent': get_random_user_agent()}
async with httpx.AsyncClient(
headers=headers, timeout=10.0, follow_redirects=True,
**httpx_proxy_kwarg(get_next_proxy()),
) as client:
resp = await request_with_backoff(
lambda: client.get('https://funpay.com/'), context='FunPay homepage'
)
if resp.status_code == 200:
tree = HTMLParser(resp.text)
game_blocks = tree.css('.promo-game-item, .game-item, .promo-game')
for block in game_blocks:
title_elem = block.css_first('.promo-game-title, .game-title, h3, header')
if not title_elem:
continue
title_text = title_elem.text().strip()
for game_key, config in targets.items():
if game_key in found_urls:
continue
if any(kw.lower() in title_text.lower() for kw in config['keywords']):
links = block.css('a')
for target_sec in config['sections']:
for link in links:
link_text = link.text().strip()
if target_sec.lower() == link_text.lower():
href = link.attributes.get('href', '')
if href:
full_url = href if href.startswith('http') else f'https://funpay.com{href}'
found_urls[game_key] = full_url
break
if game_key in found_urls:
break
except Exception as e:
logger.warning('Ошибка автопоиска FunPay URLs: %s', e)
defaults = {
'CS2': 'https://funpay.com/lots/1906/',
'Dota 2': 'https://funpay.com/lots/210/',
'TF2': 'https://funpay.com/lots/1808/',
}
for k, v in defaults.items():
if k not in found_urls:
found_urls[k] = v
return found_urls
# ==========================================
# 3. МАКЕТ ДАННЫХ И ДИНАМИЧЕСКИЙ РАСЧЕТ
# ==========================================
LISTING_FEE_PCT = 0.09
WITHDRAW_FEE_PCT = 0.03
def calc_payout_chain(sell_price_rub: float) -> tuple[float, float]:
after_listing_fee = round(sell_price_rub * (1 - LISTING_FEE_PCT), 2)
final_payout = round(after_listing_fee * (1 - WITHDRAW_FEE_PCT), 2)
return after_listing_fee, final_payout
LIQ_STEAM_HIGH_VOLUME = 20
LIQ_STEAM_LOW_VOLUME = 5
LIQ_FUNPAY_HIGH_COMPETITION = 10
LIQ_BUFF_HIGH_COMPETITION = 50
LIQUIDITY_HIGH = 'Высокая'
LIQUIDITY_MEDIUM = 'Средняя'
LIQUIDITY_LOW = 'Низкая'
LIQUIDITY_PENDING = 'Ждёт Steam'
def compute_liquidity(
steam_volume_24h: int, steam_data_loaded: bool, funpay_lots: int, buff_lots: int
) -> str:
if not steam_data_loaded:
return LIQUIDITY_PENDING
if steam_volume_24h < LIQ_STEAM_LOW_VOLUME:
return LIQUIDITY_LOW
if funpay_lots > 0:
high_competition = funpay_lots >= LIQ_FUNPAY_HIGH_COMPETITION
else:
high_competition = buff_lots >= LIQ_BUFF_HIGH_COMPETITION
if steam_volume_24h >= LIQ_STEAM_HIGH_VOLUME and not high_competition:
return LIQUIDITY_HIGH
return LIQUIDITY_MEDIUM
def process_item_math(
item_name: str,
steam_hash_name: str,
goods_id: int,
game_code: str,
app_id: int,
buff_price_cny: float,
buff_buy_max_cny: float,
funpay_price_rub: Optional[float],
funpay_offer_id: Optional[int],
steam_price_rub: float,
steam_price_is_estimated: bool,
steam_volume_24h: int,
cny_rate: float,
funpay_base_url: str,
gem_price_estimate_cny: Optional[float] = None,
history_window_hours: Optional[float] = None,
history_avg_price_rub: Optional[float] = None,
history_points_count: int = 0,
buff_lots_count: int = 0,
funpay_lots_count: int = 0,
steam_data_loaded: bool = False,
) -> dict:
if funpay_price_rub and funpay_price_rub > 0:
target_sell_price = funpay_price_rub
source_name = 'FunPay'
if funpay_offer_id:
sell_url = f'https://funpay.com/lots/offer?id={funpay_offer_id}'
else:
sell_url = f'{funpay_base_url}?query={quote_plus(item_name)}'
else:
target_sell_price = steam_price_rub
source_name = 'Steam'
sell_url = f'https://steamcommunity.com/market/listings/{app_id}/{quote(steam_hash_name)}'
if source_name == 'Steam' and steam_price_is_estimated:
source_label = f'{source_name} (оценка)'
else:
source_label = source_name
buff_url = f'https://buff.163.com/goods/{goods_id}?game={game_code}'
buff_cost_rub = round(buff_price_cny * cny_rate, 2)
amount_after_listing_fee, final_payout_rub = calc_payout_chain(target_sell_price)
net_profit_rub = round(final_payout_rub - buff_cost_rub, 2)
net_profit_pct = round(
((net_profit_rub / buff_cost_rub) * 100 if buff_cost_rub > 0 else 0.0), 2
)
if buff_price_cny > 0:
volatility_pct = round(((buff_price_cny - buff_buy_max_cny) / buff_price_cny) * 100, 2)
else:
volatility_pct = 0.0
worst_payout_rub = (buff_buy_max_cny * cny_rate) * (1 - 0.12)
worst_profit_pct = round(
(((worst_payout_rub - buff_cost_rub) / buff_cost_rub) * 100 if buff_cost_rub > 0 else 0.0), 2
)
trend_7d = round(((buff_price_cny - buff_buy_max_cny) / (buff_buy_max_cny or 1)) * 10, 2)
source_html = f'<a href="{sell_url}" target="_blank" rel="noopener noreferrer">{source_label}</a>'
if gem_price_estimate_cny:
buff_price_label = f'{buff_price_cny} ¥ (+ ~{gem_price_estimate_cny} ¥)'
else:
buff_price_label = f'{buff_price_cny} ¥'
buff_html = f'<a href="{buff_url}" target="_blank" rel="noopener noreferrer">{buff_price_label}</a>'
return {
'name': item_name,
'source_html': source_html,
'sell_price': round(target_sell_price, 2),
'buff_html': buff_html,
'buff_rub': buff_cost_rub,
'after_listing_fee': amount_after_listing_fee,
'final_payout': final_payout_rub,
'profit_rub': net_profit_rub,
'profit_pct': net_profit_pct,
'worst_profit_pct': worst_profit_pct,
'volatility_pct': volatility_pct,
'trend_7d': trend_7d,
'history_window_hours': history_window_hours,
'history_avg_price_rub': history_avg_price_rub,
'history_points_count': history_points_count,
'buff_lots': buff_lots_count,
'funpay_lots': funpay_lots_count,
'steam_volume': steam_volume_24h,
'liquidity': compute_liquidity(
steam_volume_24h, steam_data_loaded, funpay_lots_count, buff_lots_count
),
'goods_id': str(goods_id),
'app_id': app_id,
'steam_hash_name': steam_hash_name,
'steam_loaded': steam_data_loaded,
'source_label': source_label,
# Сырые ссылки (без HTML-обёртки) — нужны отдельно от source_html/
# buff_html, чтобы можно было записать их в БД (п.1 и п.5 задачи:
# синхронизация buff_url/funpay_url в purchases при повторном скане).
'buff_url': buff_url,
'sell_url': sell_url,
'sell_source': source_name, # 'FunPay' | 'Steam' — на какую площадку ведёт sell_url
}
async def fetch_steam_price(app_id: int, market_hash_name: str) -> Optional[tuple[float, int]]:
try:
async with httpx.AsyncClient(
headers={'User-Agent': get_random_user_agent(), 'Accept-Language': 'ru-RU, ru;q=0.9'},
cookies={'Steam_Language': 'russian', 'steamCountry': 'RU'},
timeout=8.0,
**httpx_proxy_kwarg(get_next_proxy()),
) as steam_client:
resp = await steam_client.get(
'https://steamcommunity.com/market/priceoverview/',
params={'appid': app_id, 'currency': 5, 'market_hash_name': market_hash_name},
)
if resp.status_code == 429:
cooldown = random.uniform(*STEAM_RATE_LIMIT_COOLDOWN_RANGE)
logger.warning('Steam API rate limit (429) на "%s" пауза %.1fc перед следующим', market_hash_name, cooldown)
await asyncio.sleep(cooldown)
return None
if resp.status_code != 200:
return None
data = resp.json()
if not data.get('success'):
return None
price_str = data.get('lowest_price') or data.get('median_price')
if not price_str:
return None
cleaned = re.sub(r'[^\d,.]', '', price_str).replace(',', '.').strip('.')
if not cleaned:
return None
price = float(cleaned)
volume_str = data.get('volume') or '0'
volume_cleaned = re.sub(r'[^\d]', '', volume_str)
volume = int(volume_cleaned) if volume_cleaned else 0
return price, volume
except Exception as e:
logger.warning('Ошибка запроса цены Steam для "%s": %s', market_hash_name, e)
return None
BUFF_MAX_RETRIES_PER_PAGE = 5
BUFF_MAX_PAGES = 3
async def fetch_all_buff_items(
client: httpx.AsyncClient, url_base: str, page_size: int = 80, max_pages: int = BUFF_MAX_PAGES
) -> list:
all_items = []
pages = list(range(1, max_pages + 1))
random.shuffle(pages)
logger.info('Buff163: порядок обхода страниц: %s', pages)
total_pages_known: Optional[int] = None
for idx, page in enumerate(pages):
if total_pages_known is not None and page > total_pages_known:
logger.debug('Buff163: стр.%s > total_page=%s, пропускаю', page, total_pages_known)
continue
retry = 0
while True:
resp = await client.get(
f'{url_base}&page_num={page}&page_size={page_size}&_={int(time.time() * 1000)}'
)
if resp.status_code != 429 and not (500 <= resp.status_code < 600):
break
retry += 1
if retry > BUFF_MAX_RETRIES_PER_PAGE:
logger.warning('Buff163: стр.%s превышен лимит повторов после %s (%s)', page, resp.status_code, BUFF_MAX_RETRIES_PER_PAGE)
return all_items
backoff = random.uniform(*REQUEST_RETRY_DELAY_RANGE) * retry + random.uniform(0, 1)
logger.warning('Buff163 %s на стр.%s, попытка %s/%s, жду %.1fc', resp.status_code, page, retry, BUFF_MAX_RETRIES_PER_PAGE, backoff)
await asyncio.sleep(backoff)
if resp.status_code != 200:
logger.warning('Buff163: стр.%s вернула %s, останавливаюсь', page, resp.status_code)
break
data = resp.json()
if data.get('code') != 'OK':
if data.get('code') == 'Login Required':
logger.warning('Buff163: сессия невалидна/протухла (Login Required) на стр.%s', page)
else:
logger.warning('Buff163: стр.%s, code=%s, останавливаюсь', page, data.get('code'))
break
payload = data.get('data', {}) or {}
if total_pages_known is None and payload.get('total_page'):
try:
total_pages_known = int(payload['total_page'])
except (TypeError, ValueError):
pass
items = payload.get('items', [])
if items:
all_items.extend(items)
else:
logger.debug('Buff163: стр.%s пустая', page)
remaining = [p for p in pages[idx + 1:] if total_pages_known is None or p <= total_pages_known]
if remaining:
await asyncio.sleep(random.uniform(*BUFF_DELAY_RANGE) + random.uniform(*BUFF_PAGE_EXTRA_JITTER_RANGE))
return all_items
# ==========================================
# 3a. ФОНОВАЯ STEAM-ОЧЕРЕДЬ
# ==========================================
STEAM_PRIORITY_URGENT = 0
STEAM_PRIORITY_ESTIMATED = 1
STEAM_PRIORITY_VOLUME = 2
STEAM_QUEUE_MAX_RETRIES = 2
class SteamFetchQueue:
def __init__(self) -> None:
self._pq: asyncio.PriorityQueue = asyncio.PriorityQueue()
self._seq = itertools.count()
self._pending: Dict[str, tuple[int, int]] = {}
self._jobs: Dict[str, dict] = {}
self.in_progress: Optional[str] = None
self.failed: set = set()
def __len__(self) -> int:
return len(self._pending)
def is_queued(self, goods_id: str) -> bool:
return goods_id in self._pending
def priority_of(self, goods_id: str) -> Optional[int]:
entry = self._pending.get(goods_id)
return entry[0] if entry else None
def put(
self, goods_id: str, app_id: int, hash_name: str, item_name: str,
priority: int = STEAM_PRIORITY_VOLUME,
) -> str:
goods_id = str(goods_id)
if self.in_progress == goods_id:
return 'in_progress'
if steam_cache_is_fresh(goods_id):
return 'loaded'
current = self._pending.get(goods_id)
if current is not None and current[0] <= priority:
return 'already_queued'
seq = next(self._seq)
job = self._jobs.get(goods_id, {'retries': 0})
job.update(app_id=app_id, hash_name=hash_name, item_name=item_name, priority=priority)
self._jobs[goods_id] = job
self._pending[goods_id] = (priority, seq)
self.failed.discard(goods_id)
self._pq.put_nowait((priority, seq, goods_id))
return 'bumped' if current is not None else 'queued'
async def get(self) -> tuple[str, dict]:
while True:
priority, seq, goods_id = await self._pq.get()
if self._pending.get(goods_id) != (priority, seq):
continue
del self._pending[goods_id]
return goods_id, self._jobs[goods_id]
def clear(self) -> None:
self._pending.clear()
self._jobs = {k: v for k, v in self._jobs.items() if k == self.in_progress}
self.failed.clear()
while not self._pq.empty():
self._pq.get_nowait()
steam_queue = SteamFetchQueue()
ROW_REGISTRY: Dict[str, dict] = {}
ROW_CONTEXTS: Dict[str, dict] = {}
# Набор goods_id, уже добавленных в "Избранное" — обновляется в начале каждого
# refresh_data() и точечно при добавлении/удалении покупки. Проверяется на
# каждой пересборке строки (build_row), чтобы НЕ дёргать sync_purchase_market_links
# (лишний UPDATE в SQLite) для сотен обычных, неотслеживаемых товаров скана.
_TRACKED_PURCHASE_IDS: set = set()
def steam_status_label(goods_id: str, steam_loaded: bool) -> str:
if steam_queue.in_progress == goods_id:
return 'Загрузка Steam...'
if steam_loaded:
return 'Steam загружен'
if goods_id in steam_queue.failed:
return 'Ошибка (повторить)'
if steam_queue.priority_of(goods_id) == STEAM_PRIORITY_URGENT:
return 'Следующий в очереди'
return 'В начало очереди'
def build_row(ctx: dict) -> dict:
row = process_item_math(**ctx)
row['queue_action'] = steam_status_label(row['goods_id'], row['steam_loaded'])
row['buy_action'] = 'В покупки'
if row['goods_id'] in _TRACKED_PURCHASE_IDS:
sync_purchase_market_links(
row['goods_id'],
buff_url=row.get('buff_url'),
funpay_url=row.get('sell_url') if row.get('sell_source') == 'FunPay' else None,
funpay_matched=(row.get('sell_source') == 'FunPay'),
steam_data_loaded=bool(row.get('steam_loaded')),
)
return row
async def process_one_item(
item: dict,
client: httpx.AsyncClient,
funpay_offers: list,
fx_rates: Dict[str, float],
game_code: str,
app_id: int,
cny_rate: float,
funpay_base: str,
funpay_session: AsyncSession,
) -> Optional[dict]:
goods_id = item.get('id')
market_hash_name = item.get('market_hash_name')
localized_name = item.get('name') or 'Unknown Item'
item_name = market_hash_name or localized_name
steam_hash_name = market_hash_name or localized_name
if not market_hash_name:
logger.warning('Товар goods_id=%s без market_hash_name использую локализованное название "%s"', goods_id, localized_name)
elif has_cjk(localized_name) and localized_name != market_hash_name:
logger.debug('goods_id=%s: localized_name на китайском ("%s"), использую market_hash_name ("%s")', goods_id, localized_name, market_hash_name)
tags = item.get('goods_info', {}).get('info', {}).get('tags', {}) or {}
target_rarity = (tags.get('rarity', {}).get('localized_name') or '').strip().lower() or None
item_type = (tags.get('type', {}).get('localized_name') or '').strip().lower()
is_bundle = item_type == 'bundle'
if is_bundle:
logger.debug('"%s" бандл (Buff163 tags.type=Bundle)', item_name)
target_quality = extract_target_quality(item_name)
buff_price_cny = float(item.get('sell_min_price', 0.0))
buff_buy_max_cny = float(item.get('buy_max_price', 0.0))
buff_cost_rub = buff_price_cny * cny_rate
volume = int(item.get('sell_num', 0))
history_rows, history_window_hours = get_recent_price_history(goods_id)
history_prices = [row['price_rub'] for row in history_rows]
history_avg_price_rub = round(sum(history_prices) / len(history_prices), 2) if history_prices else None
history_points_count = len(history_prices)
funpay_match = get_min_funpay_offer(
item_name,
funpay_offers,
fx_rates,
strict=is_bundle,
target_rarity=target_rarity,
target_quality=target_quality,
)
if funpay_match:
funpay_price_rub, funpay_offer_id, gem_hint, funpay_lots_count = funpay_match
gem_price_estimate_cny: Optional[float] = None
if gem_hint:
gem_price_estimate_cny = round(buff_price_cny * GEM_PRICE_ESTIMATE_PCT, 2)
if REFINE_FUNPAY_PRICE_VIA_OFFER_PAGE:
details = await fetch_funpay_offer_details(funpay_offer_id, fx_rates, funpay_session)
if details is not None:
refined_price, offer_fields_details = details
funpay_price_rub = refined_price
else:
cached = cache_get_offer(funpay_offer_id)
if cached is not None:
funpay_price_rub = cached['price_rub']
logger.debug('Лот %s: живой запрос цены не удался, беру последнюю сохранённую (%.2f P)', funpay_offer_id, funpay_price_rub)
cache_upsert_offer(funpay_offer_id, item_name, funpay_price_rub, 'matched')
await asyncio.sleep(random.uniform(*FUNPAY_DELAY_RANGE))
else:
funpay_price_rub, funpay_offer_id = None, None
gem_price_estimate_cny = None
funpay_lots_count = 0
target_lower = item_name.lower()
loosely_present = [offer for offer in funpay_offers if offer.title and target_lower in offer.title.lower()]
if loosely_present:
logger.warning('"%s": лот(ы) ЕСТЬ в загруженном каталоге (id: %s), но не прошли фильтры', item_name, [getattr(o, 'id', '?') for o in loosely_present])
steam_price_is_estimated = False
real_steam_volume = 0
steam_data_loaded = False
estimated_steam_rub = 0.0
steam_cached = steam_cache_get(goods_id)
if is_cache_fresh(steam_cached, ttl=STEAM_PRICE_TTL_SECONDS):
real_steam_volume = steam_cached['volume_24h'] or 0
steam_data_loaded = True
if funpay_price_rub is None:
estimated_steam_rub = steam_cached['price_rub']
logger.debug('"%s": Steam из кэша очереди (%.2f Р, объём %s)', item_name, steam_cached['price_rub'] or 0, real_steam_volume)
elif funpay_price_rub is None:
cached_scan = cache_get_scan_result(goods_id)
if is_cache_fresh(cached_scan, ttl=STEAM_CACHE_TTL_SECONDS) and cached_scan['source'] == 'steam':
estimated_steam_rub = cached_scan['price_rub']
real_steam_volume = cached_scan['steam_volume'] or 0
steam_data_loaded = True
logger.debug('"%s": цена Steam из кэша scan_results (%.2f Р, объём %s)', item_name, estimated_steam_rub, real_steam_volume)
else:
estimated_steam_rub = round(buff_cost_rub * 1.35, 2)
steam_price_is_estimated = True
if funpay_price_rub is not None:
cache_upsert_scan_result(goods_id, item_name, 'funpay', funpay_price_rub, 'matched')
else:
source = 'steam_estimated' if steam_price_is_estimated else 'steam'
cache_upsert_scan_result(goods_id, item_name, source, estimated_steam_rub, 'no_match', steam_volume=real_steam_volume)
ctx = dict(
item_name=item_name,
steam_hash_name=steam_hash_name,
goods_id=goods_id,
game_code=game_code,
app_id=app_id,
buff_price_cny=buff_price_cny,
buff_buy_max_cny=buff_buy_max_cny,
gem_price_estimate_cny=gem_price_estimate_cny,
funpay_price_rub=funpay_price_rub,
funpay_offer_id=funpay_offer_id,
steam_price_rub=estimated_steam_rub,
steam_price_is_estimated=steam_price_is_estimated,
buff_lots_count=volume,
funpay_lots_count=funpay_lots_count,
steam_volume_24h=real_steam_volume,
cny_rate=cny_rate,
funpay_base_url=funpay_base,
history_window_hours=round(history_window_hours, 2),
history_avg_price_rub=history_avg_price_rub,
history_points_count=history_points_count,
steam_data_loaded=steam_data_loaded,
)
row = build_row(ctx)
ROW_CONTEXTS[row['goods_id']] = ctx
ROW_REGISTRY[row['goods_id']] = row
return row
def passes_lots_filters(
row: dict, lots_source: str, min_sales: int, min_steam_sales: int
) -> bool:
steam_known = bool(row.get('steam_loaded', True))
if min_steam_sales > 0 and steam_known and row.get('steam_volume', 0) < min_steam_sales:
return False
if lots_source == 'FunPay':
return row.get('funpay_lots', 0) >= min_sales
if lots_source == 'Steam':
return (not steam_known) or row.get('steam_volume', 0) >= min_sales
if lots_source == 'Все вместе':
total_lots = row.get('buff_lots', 0) + row.get('funpay_lots', 0) + row.get('steam_volume', 0)
return total_lots >= min_sales or not steam_known
return True
async def fetch_market_data(
game: str, cny_rate: float, min_rub: float, min_sales: int,
lots_source: str = 'Buff',
min_steam_sales: int = 0,
on_batch: Optional[Callable[[List[dict]], None]] = None,
) -> List[dict]:
clean_session = BUFF_SESSION.encode('ascii', 'ignore').decode('ascii').strip()
if not clean_session:
logger.error('Вставьте корректный BUFF_SESSION!')
return []
init_db()
game_code = BUFF_GAME_CODES.get(game, 'csgo')
app_id = STEAM_APP_IDS.get(game, 730)
url_base = f'https://buff.163.com/api/market/goods?game={game_code}'
headers = build_buff_headers(game_code, clean_session)
processed: List[dict] = []
funpay_base = funpay_urls.get(game, 'https://funpay.com/lots/1906/')
proxy_url = get_next_proxy()
if proxy_url:
logger.info('Использую прокси для этого прогона: %s', proxy_url)
funpay_session = AsyncSession(
impersonate='chrome120',
cookies={'locale': 'ru', 'cy': 'RUB'},
headers={'Accept-Language': 'ru-RU,ru;q=0.9,en;q=0.1'},
**curl_cffi_proxy_kwarg(proxy_url),
)
funpay_offers = await fetch_funpay_offers(funpay_base, funpay_session)
logger.info('FunPay (%s): загружено лотов: %s', funpay_base, len(funpay_offers))
fx_rates = await fetch_fx_rates_to_rub()
try:
async with httpx.AsyncClient(
headers=headers, timeout=12.0, follow_redirects=True,
**httpx_proxy_kwarg(proxy_url),
) as client:
items = await fetch_all_buff_items(client, url_base)
saved = record_buff_snapshot(items, cny_rate, game)
logger.info('Buff163: срез цен сохранён в SQLite (%s товаров)', saved)
qualifying_items = [
item for item in items
if float(item.get('sell_min_price', 0.0)) * cny_rate > min_rub
and (lots_source != 'Buff' or int(item.get('sell_num', 0)) >= min_sales)
]
total = len(qualifying_items)
logger.info('Buff163: %s товаров прошли порог из %s, обрабатываю батчами по %s', total, len(items), BATCH_SIZE)
for batch_num, batch in enumerate(chunked(qualifying_items, BATCH_SIZE)):
for item in batch:
try:
result = await process_one_item(
item, client=client, funpay_offers=funpay_offers,
fx_rates=fx_rates, game_code=game_code, app_id=app_id,
cny_rate=cny_rate, funpay_base=funpay_base, funpay_session=funpay_session,
)
except Exception as e:
logger.exception('Ошибка обработки товара "%s" (goods_id=%s): %s', item.get('name'), item.get('id'), e)
continue
if result is not None and passes_lots_filters(result, lots_source, min_sales, min_steam_sales):
processed.append(result)
done = min((batch_num + 1) * BATCH_SIZE, total)
logger.info('Батч %s обработан: %s/%s товаров', batch_num, done, total)
if on_batch is not None:
try:
on_batch(list(processed))
except Exception:
logger.exception('Ошибка в колбэке on_batch (батч %s)', batch_num)
if done < total:
await asyncio.sleep(random.uniform(*INTER_BATCH_DELAY_RANGE))
except Exception as e:
logger.exception('Ошибка запроса к Buff163: %s', e)
finally:
await funpay_session.close()
return processed
# ==========================================
# 4. ИНТЕРФЕЙС NICEGUI + AG GRID
# ==========================================
ENABLE_AUTO_REFRESH = True
AUTO_REFRESH_INTERVAL_MINUTES = 60
current_game = 'Dota 2'
cny_rub_rate = 12.5
funpay_urls: Dict[str, str] = {}
LIQUIDITY_COMPARATOR_JS = '''(a, b) => {
const rank = (v) => ({
'Вы': 3,
'Ср': 2,
'Ни': 1
}[(v || '').slice(0, 2)] || 0);
return rank(a) - rank(b);
}'''
def age_cell_class_js(ratio_field: str) -> str:
"""
JS-функция для ':cellClass' (не ':cellClassRules'!) -- этот враппер NiceGUI
надёжно подставляет как raw JS только ОДНУ строку на colon-префиксный ключ
(см. везде в файле ':comparator', ':getRowId', ':onFirstDataRendered' --
всегда одна строка-функция, никогда вложенный словарь строк). Поэтому
вместо нескольких именованных правил cellClassRules -- одна функция,
которая сама решает, какой класс вернуть, на основе поля-компаньона
ratio_field = age_hours / history_window_hours (см. _age_ratio, п.4 задачи).
"""
return (
"params => {"
f"const r = params.data.{ratio_field};"
"if (r == null) return 'text-gray-500 italic';"
"if (r <= 1.0) return 'bg-green-900/40 text-green-300';"
"if (r <= 2.0) return 'bg-yellow-900/40 text-yellow-300';"
"return 'bg-red-900/50 text-red-300 font-bold';"
"}"
)
grid_options = {
'enableCellTextSelection': True,
'ensureDomOrder': True,
':getRowId': '(params) => String(params.data.goods_id)',
'columnDefs': [
{'field': 'name', 'headerName': 'Предмет', 'flex': 2, 'filter': 'agTextColumnFilter'},
{'field': 'source_html', 'headerName': 'Источник', 'flex': 1, 'filter': 'agTextColumnFilter'},
{'field': 'sell_price', 'headerName': 'Цена продажи (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'buff_html', 'headerName': 'Buff (Y)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'buff_rub', 'headerName': 'Закуп (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'final_payout', 'headerName': 'На карту (-3%)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{
'field': 'profit_rub', 'headerName': 'Профит (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter',
'cellClassRules': {
'bg-green-900/50 text-green-300 font-bold': 'x > 0',
'bg-red-900/50 text-red-300 font-bold': 'x <= 0',
},
},
{
'field': 'profit_pct', 'headerName': 'Профит %', 'flex': 1, 'filter': 'agNumberColumnFilter',
'cellClassRules': {
'bg-green-900/50 text-green-300 font-bold': 'x >= 15.0',
'bg-yellow-900/40 text-yellow-300': 'x >= 5.0 && x < 15.0',
'bg-red-900/50 text-red-300': 'x < 5.0',
},
},
{
'field': 'worst_profit_pct', 'headerName': 'Худший (7д)', 'flex': 1, 'filter': 'agNumberColumnFilter',
'cellClassRules': {
'text-green-400 font-bold': 'x > 5.0',
'text-red-400 font-bold': 'x < 0.0',
'text-yellow-400': 'x >= 0.0 && x <= 5.0',
},
},
{
'field': 'volatility_pct', 'headerName': 'Волатильность %', 'flex': 1, 'filter': 'agNumberColumnFilter',
'cellClassRules': {
'text-green-400 font-bold': 'x >= 0 && x <= 8.0',
'text-yellow-400 font-bold': 'x > 8.0 && x <= 15.0',
'text-red-400 font-bold': 'x > 15.0',
},
},
{
'field': 'trend_7d', 'headerName': 'Тренд 7д %', 'flex': 1, 'filter': 'agNumberColumnFilter',
'cellClassRules': {
'text-green-400 font-bold': 'x > 0',
'text-red-400 font-bold': 'x < 0',
},
},
{
'field': 'liquidity', 'headerName': 'Индекс ликвидности', 'flex': 1, 'minWidth': 140,
'filter': 'agTextColumnFilter', ':comparator': LIQUIDITY_COMPARATOR_JS,
'cellClassRules': {
'bg-green-900/50 text-green-300 font-bold': f'x === "{LIQUIDITY_HIGH}"',
'bg-yellow-900/40 text-yellow-300 font-bold': f'x === "{LIQUIDITY_MEDIUM}"',
'bg-red-900/50 text-red-300 font-bold': f'x === "{LIQUIDITY_LOW}"',
'text-gray-400 italic': f'x === "{LIQUIDITY_PENDING}"',
},
},
{'field': 'history_window_hours', 'headerName': 'Окно анализа (ч)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'history_avg_price_rub', 'headerName': 'Средняя цена в окне (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'buff_lots', 'headerName': 'Лоты Buff', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'funpay_lots', 'headerName': 'Лоты FunPay', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'steam_volume', 'headerName': 'Продажи Steam (24ч)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{
'field': 'queue_action', 'headerName': 'Steam-очередь', 'pinned': 'right', 'width': 190,
'sortable': False, 'filter': False, 'floatingFilter': False, 'cellClass': 'cursor-pointer text-amber-300 font-semibold',
},
{
'field': 'buy_action', 'headerName': 'Покупка', 'pinned': 'right', 'width': 140,
'sortable': False, 'filter': False, 'floatingFilter': False, 'cellClass': 'cursor-pointer text-sky-300 font-semibold',
},
],
'defaultColDef': {
'resizable': True, 'sortable': True, 'filter': True, 'floatingFilter': True,
'wrapHeaderText': True, 'autoHeaderHeight': True, 'minWidth': 90,
},
'rowData': [],
# Автоматический пересчёт ширины колонок под содержимое — при первой
# отрисовке и при КАЖДОМ обновлении rowData (батчи сканирования, фоновые
# правки от Steam-очереди через applyTransaction). Без этого flex просто
# делит место поровну и не учитывает, влезает ли реальный текст/число.
':onFirstDataRendered': 'params => params.api.autoSizeAllColumns()',
':onRowDataUpdated': 'params => params.api.autoSizeAllColumns()',
':onGridSizeChanged': 'params => params.api.autoSizeAllColumns()',
}
favorites_grid_options = {
'enableCellTextSelection': True,
'ensureDomOrder': True,
':getRowId': '(params) => String(params.data.id)',
'columnDefs': [
{'field': 'purchased_at_str', 'headerName': 'Дата покупки', 'flex': 1, 'filter': 'agTextColumnFilter'},
{'field': 'item_name_html', 'headerName': 'Предмет', 'flex': 2, 'filter': 'agTextColumnFilter'},
{'field': 'game', 'headerName': 'Игра', 'flex': 1, 'filter': 'agTextColumnFilter'},
{
'field': 'buy_price_rub', 'headerName': 'Цена покупки (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter',
# Редактирование прямо в ячейке (п.3 задачи) — сохранение и
# пересчёт профита см. on_favorites_cell_value_changed.
'editable': True, ':valueParser': 'params => Number(params.newValue)',
},
{
'field': 'sell_price_rub', 'headerName': 'Моя цена продажи (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter',
'editable': True, ':valueParser': 'params => Number(params.newValue)',
# Сравнение с текущим минимумом FunPay (п.3 задачи): зелёная
# плашка — моя цена конкурентна (не выше рыночного минимума),
# красная — моя цена завышена относительно него. ':cellClass'
# (одна JS-функция, а не 'cellClassRules' с шаблоном 'x') — нужен
# доступ к ДРУГОМУ полю той же строки (funpay_min_price_rub), а не
# только к значению своей же ячейки.
':cellClass': (
"params => {"
"const min = params.data.funpay_min_price_rub;"
"if (min == null || params.value == null) return '';"
"return params.value <= min "
"? 'bg-green-900/50 text-green-300 font-bold' "
": 'bg-red-900/50 text-red-300 font-bold';"
"}"
),
},
{'field': 'current_price_rub', 'headerName': 'Рыночная цена (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'funpay_min_price_rub', 'headerName': 'Мин. FunPay (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{'field': 'price_source', 'headerName': 'Источник цены', 'flex': 1, 'filter': 'agTextColumnFilter'},
{'field': 'buff_now_rub', 'headerName': 'Buff сейчас (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter'},
{
'field': 'profit_rub', 'headerName': 'Чистый профит (Р)', 'flex': 1, 'filter': 'agNumberColumnFilter',
'cellClassRules': {
'bg-green-900/50 text-green-300 font-bold': 'x != null && x > 0',
'bg-red-900/50 text-red-300 font-bold': 'x != null && x <= 0',
},
},
{
'field': 'profit_pct', 'headerName': 'Профит %', 'flex': 1, 'filter': 'agNumberColumnFilter',
'cellClassRules': {
'bg-green-900/50 text-green-300 font-bold': 'x != null && x > 0',
'bg-red-900/50 text-red-300 font-bold': 'x != null && x <= 0',
},
},
{'field': 'funpay_html', 'headerName': 'FunPay', 'flex': 1, 'filter': False, 'sortable': False},
{
'field': 'price_updated_str', 'headerName': 'Цена обновлена', 'flex': 1, 'filter': 'agTextColumnFilter',
':cellClass': age_cell_class_js('price_age_ratio'),
},
{
'field': 'buff_updated_str', 'headerName': 'Buff обновлён', 'flex': 1, 'filter': 'agTextColumnFilter',
':cellClass': age_cell_class_js('buff_age_ratio'),
},
{
'field': 'funpay_updated_str', 'headerName': 'FunPay обновлён', 'flex': 1, 'filter': 'agTextColumnFilter',
':cellClass': age_cell_class_js('funpay_age_ratio'),
},
{
'field': 'steam_updated_str', 'headerName': 'Steam обновлён', 'flex': 1, 'filter': 'agTextColumnFilter',
':cellClass': age_cell_class_js('steam_age_ratio'),
},
{
'field': 'delete_action', 'headerName': '', 'pinned': 'right', 'width': 100,
'sortable': False, 'filter': False, 'floatingFilter': False, 'cellClass': 'cursor-pointer text-red-300',
},
],
'defaultColDef': {
'resizable': True, 'sortable': True, 'filter': True, 'floatingFilter': True,
'wrapHeaderText': True, 'autoHeaderHeight': True, 'minWidth': 90,
},
'rowData': [],
# Важно: не запускаем autoSizeAllColumns на скрытом grid. Когда вкладка
# ещё скрыта, clientWidth == 0, и AG Grid может присвоить обычным колонкам
# ширину 0. После открытия вкладки они тогда визуально "исчезают" до
# следующего resize. Проверяем реальную ширину контейнера.
':onFirstDataRendered': '''params => {
const width = params.eGridDiv?.getBoundingClientRect().width || 0;
if (width > 100) requestAnimationFrame(() => params.api.autoSizeAllColumns());
}''',
':onRowDataUpdated': '''params => {
const width = params.eGridDiv?.getBoundingClientRect().width || 0;
if (width > 100) requestAnimationFrame(() => params.api.autoSizeAllColumns());
}''',
':onGridSizeChanged': '''params => {
const width = params.eGridDiv?.getBoundingClientRect().width || 0;
if (width > 100) requestAnimationFrame(() => params.api.autoSizeAllColumns());
}''',
}
SOURCE_LABELS = {'funpay': 'FunPay', 'steam': 'Steam', 'steam_estimated': 'Steam (оценка)'}
init_db()
ui.dark_mode().enable()
with ui.column().classes('w-full p-6 gap-4'):
ui.label('Арбитражный сканер (Buff163 / FunPay / Steam)').classes('text-2xl font-bold')
with ui.tabs().classes('w-full') as tabs:
scanner_tab = ui.tab('scanner', label='Сканер арбитража')
favorites_tab = ui.tab('favorites', label='Мои покупки / Избранное')
with ui.tab_panels(tabs, value=scanner_tab).classes('w-full bg-transparent'):
with ui.tab_panel(scanner_tab).classes('p-0 gap-4'):
with ui.row().classes('w-full items-center gap-4 bg-gray-800 p-4 rounded-lg flex-wrap'):
game_select = ui.toggle(['CS2', 'Dota 2', 'TF2'], value='Dota 2').props('clearable=False color=blue')
rate_label = ui.label(f'Курс CNY: {cny_rub_rate} Р').classes('text-lg font-semibold text-yellow-400')
min_price_input = ui.number(label='Мин. цена (Р)', value=500.0, min=0, step=100).classes('w-32 bg-gray-700 text-white').props('dense outlined dark')
min_sales_input = ui.number(label='Мин. лотов', value=10, min=0, step=5).classes('w-32 bg-gray-700 text-white').props('dense outlined dark')
lots_source_select = ui.select(['Buff', 'FunPay', 'Steam', 'Все вместе'], value='Buff', label='Источник лотов').classes('w-36 bg-gray-700 text-white').props('dense outlined dark')
min_steam_sales_input = ui.number(label='Мин. продаж (Steam)', value=0, min=0, step=5).classes('w-40 bg-gray-700 text-white').props('dense outlined dark')
with ui.row().classes('items-center gap-2 ml-auto'):
ui.label('Обновление (мин):').classes('text-sm text-gray-300')
interval_input = ui.number(value=AUTO_REFRESH_INTERVAL_MINUTES, min=5, step=5).classes('w-20 bg-gray-700 text-white').props('dense outlined dark')
auto_refresh_switch = ui.switch('Авто', value=ENABLE_AUTO_REFRESH).classes('text-sm text-gray-300')
refresh_btn = ui.button('Обновить', icon='refresh').props('outline')
with ui.row().classes('w-full items-center gap-4'):
search_input = ui.input(placeholder='Поиск по названию...').classes('w-64 mb-2')
queue_status_label = ui.label('Steam-очередь: пусто').classes('text-sm text-gray-400')
grid = ui.aggrid(grid_options, html_columns=[1, 3]).classes('w-full h-[550px]')
search_input.on('update:model-value', lambda e: grid.run_grid_method('setGridOption', 'quickFilterText', e.value))
with ui.tab_panel(favorites_tab).classes('p-0 gap-4'):
with ui.row().classes('w-full items-center gap-4 bg-gray-800 p-4 rounded-lg flex-wrap'):
favorites_summary_label = ui.label('Покупок: 0').classes('text-lg font-semibold')
favorites_refresh_btn = ui.button('Обновить', icon='refresh').props('outline').classes('ml-auto')
ui.label('Текущая цена — цена продажи из последнего скана; профит считается с теми же комиссиями, что и в сканере.').classes('text-sm text-gray-400')
favorites_grid = ui.aggrid(favorites_grid_options, html_columns=[1, 11]).classes('w-full h-[550px]')
_buy_dialog_row: dict = {}
with ui.dialog() as buy_dialog, ui.card().classes('min-w-[380px]'):
buy_dialog_title = ui.label('').classes('text-lg font-bold')
buy_dialog_hint = ui.label('').classes('text-sm text-gray-400')
buy_price_input = ui.number(label='Цена закупки (Р)', min=0, step=1, format='%.2f').classes('w-full')
sell_price_input = ui.number(label='Моя цена продажи (Р)', min=0, step=1, format='%.2f').classes('w-full')
with ui.row().classes('w-full justify-end gap-2'):
ui.button('Отмена', on_click=buy_dialog.close).props('flat')
buy_save_btn = ui.button('Сохранить', icon='save')
_delete_target: dict = {}
with ui.dialog() as delete_dialog, ui.card():
delete_dialog_label = ui.label('')
with ui.row().classes('w-full justify-end gap-2'):
ui.button('Отмена', on_click=delete_dialog.close).props('flat')
delete_confirm_btn = ui.button('Удалить', color='red')
_refresh_in_progress = False
current_scan_filters: Dict[str, object] = {'lots_source': 'Buff', 'min_sales': 0, 'min_steam_sales': 0}
def _visible_rows(rows: List[dict]) -> List[dict]:
return [r for r in rows if not r.get('filtered_out')]
def _enqueue_rows_for_steam(rows: List[dict]) -> None:
for row in rows:
if row.get('steam_loaded') or row.get('filtered_out'):
continue
priority = STEAM_PRIORITY_ESTIMATED if row.get('source_label', '').startswith('Steam') else STEAM_PRIORITY_VOLUME
steam_queue.put(
row['goods_id'], row['app_id'], row['steam_hash_name'], row['name'], priority=priority
)
row['queue_action'] = steam_status_label(row['goods_id'], False)
def _push_row_update(row: dict) -> None:
try:
grid.run_grid_method('applyTransaction', {'update': [row]})
grid.run_grid_method('autoSizeAllColumns')
except Exception:
logger.debug('applyTransaction (update) не удался для %s', row.get('goods_id'))
def _remove_row(row: dict) -> None:
row['filtered_out'] = True
rows = grid.options.get('rowData') or []
grid.options['rowData'] = [r for r in rows if r.get('goods_id') != row.get('goods_id')]
try:
grid.run_grid_method('applyTransaction', {'remove': [{'goods_id': row['goods_id']}]})
except Exception:
logger.debug('applyTransaction (remove) не удался для %s', row.get('goods_id'))
def _set_row_status(goods_id: str) -> None:
row = ROW_REGISTRY.get(goods_id)
if row is None or row.get('filtered_out'):
return
row['queue_action'] = steam_status_label(goods_id, row.get('steam_loaded', False))
_push_row_update(row)
def _apply_steam_result(goods_id: str, price_rub: float, volume_24h: int) -> None:
ctx = ROW_CONTEXTS.get(goods_id)
row = ROW_REGISTRY.get(goods_id)
if ctx is None or row is None:
return
ctx['steam_volume_24h'] = volume_24h
ctx['steam_data_loaded'] = True
if not ctx.get('funpay_price_rub'):
ctx['steam_price_rub'] = price_rub
ctx['steam_price_is_estimated'] = False
cache_upsert_scan_result(goods_id, ctx['item_name'], 'steam', price_rub, 'no_match', steam_volume=volume_24h)
was_filtered = row.get('filtered_out', False)
new_row = build_row(ctx)
row.clear()
row.update(new_row)
if was_filtered:
row['filtered_out'] = True
return
if not passes_lots_filters(
row,
str(current_scan_filters['lots_source']),
int(current_scan_filters['min_sales']),
int(current_scan_filters['min_steam_sales']),
):
logger.info('"%s": после загрузки Steam не проходит фильтр лотов, убираю', row.get('name'))
_remove_row(row)
return
_push_row_update(row)
async def steam_background_worker() -> None:
logger.info('Фоновый Steam-воркер запущен')
while not grid.is_deleted:
try:
goods_id, job = await asyncio.wait_for(steam_queue.get(), timeout=5.0)
except asyncio.TimeoutError:
continue
steam_queue.in_progress = goods_id
_set_row_status(goods_id)
try:
result = await fetch_steam_price(job['app_id'], job['hash_name'])
if result is not None:
price_rub, volume_24h = result
steam_cache_upsert(goods_id, job['hash_name'], job['app_id'], price_rub, volume_24h)
steam_queue.in_progress = None
job['retries'] = 0
_apply_steam_result(goods_id, price_rub, volume_24h)
logger.info('Steam-очередь: "%s" — %.2f Р, продаж за 24ч: %s (осталось %s)', job['item_name'], price_rub, volume_24h, len(steam_queue))
if goods_id in purchased_goods_ids():
refresh_favorites()
else:
steam_queue.in_progress = None
if job['retries'] < STEAM_QUEUE_MAX_RETRIES:
job['retries'] += 1
steam_queue.put(goods_id, job['app_id'], job['hash_name'], job['item_name'], job['priority'])
else:
steam_queue.failed.add(goods_id)
logger.warning('Steam-очередь: "%s" не удалось получить цену', job['item_name'])
_set_row_status(goods_id)
except Exception:
steam_queue.in_progress = None
logger.exception('Steam-воркер: ошибка на goods_id=%s', goods_id)
await asyncio.sleep(random.uniform(*STEAM_QUEUE_DELAY_RANGE))
logger.info('Фоновый Steam-воркер остановлен (страница закрыта)')
_steam_worker_task: Optional[asyncio.Task] = None
def ensure_steam_worker() -> None:
global _steam_worker_task
if _steam_worker_task is None or _steam_worker_task.done():
_steam_worker_task = background_tasks.create(steam_background_worker(), name='steam_background_worker')
def update_queue_status_label() -> None:
current = ROW_REGISTRY.get(steam_queue.in_progress or '')
current_txt = f' сейчас: {current["name"]}' if current else ''
queue_status_label.set_text(
f'Steam-очередь: {len(steam_queue)} в ожидании{current_txt}'
if len(steam_queue) or current else 'Steam-очередь: пусто'
)
def prioritize_row(row: dict) -> None:
goods_id = str(row.get('goods_id', ''))
if not goods_id:
return
if row.get('steam_loaded'):
ui.notify('Steam-данные по этому предмету уже загружены', type='info')
return
status = steam_queue.put(
goods_id, row['app_id'], row['steam_hash_name'], row['name'], STEAM_PRIORITY_URGENT,
)
messages = {
'bumped': ('Поднят в начало очереди', 'positive'),
'queued': ('Добавлен в начало очереди', 'positive'),
'already_queued': ('Уже первый в очереди', 'info'),
'in_progress': ('Уже загружается прямо сейчас', 'info'),
'loaded': ('Steam-цена уже есть в кэше, обновите скан', 'info'),
}
text, kind = messages.get(status, (status, 'info'))
ui.notify(f'{row["name"]}: {text}', type=kind)
_set_row_status(goods_id)
def open_buy_dialog(row: dict) -> None:
_buy_dialog_row.clear()
_buy_dialog_row.update(row)
buy_dialog_title.set_text(f'Добавить в покупки: {row.get("name", "")}')
buy_dialog_hint.set_text(f'Игра: {current_game} | закуп Buff сейчас: {row.get("buff_rub", 0)} Р')
buy_price_input.value = row.get('buff_rub')
# Предзаполняем актуальной ценой продажи из сканера (п.3 задачи) — но
# поле остаётся редактируемым, пользователь может указать свою цену листинга.
sell_price_input.value = row.get('sell_price')
buy_dialog.open()
def save_purchase() -> None:
row = dict(_buy_dialog_row)
buy_price = buy_price_input.value
sell_price = sell_price_input.value
if not row or buy_price is None or float(buy_price) <= 0:
ui.notify('Укажите цену закупки больше 0', type='warning')
return
if sell_price is None or float(sell_price) <= 0:
ui.notify('Укажите цену продажи больше 0', type='warning')
return
funpay_url = row.get('sell_url') if row.get('sell_source') == 'FunPay' else None
purchase_add(
row['goods_id'], row['name'], current_game, float(buy_price),
sell_price_rub=float(sell_price),
buff_url=row.get('buff_url'),
funpay_url=funpay_url,
)
_TRACKED_PURCHASE_IDS.add(str(row['goods_id']))
buy_dialog.close()
ui.notify(f'{row["name"]} добавлен в покупки за {float(buy_price):.2f} Р', type='positive')
refresh_favorites()
buy_save_btn.on_click(save_purchase)
def on_scanner_cell_clicked(e) -> None:
args = e.args or {}
col_id = args.get('colId')
data = args.get('data') or {}
if not data:
return
row = ROW_REGISTRY.get(str(data.get('goods_id')), data)
if col_id == 'queue_action':
prioritize_row(row)
elif col_id == 'buy_action':
open_buy_dialog(row)
grid.on('cellClicked', on_scanner_cell_clicked)
def _fmt_ts(ts: Optional[float]) -> str:
return datetime.fromtimestamp(ts).strftime('%d.%m.%Y %H:%M') if ts else '-'
def _age_ratio(ts: Optional[float], window_hours: Optional[float]) -> Optional[float]:
"""
Возраст цены (в долях "окна анализа") — 1.0 = ровно на границе окна,
>1.0 = цена старше, чем окно, за которое алгоритм считает её актуальной
(см. п.4 задачи: history_window_hours). None, если нет ни метки времени,
ни окна — тогда красить ячейку не во что, оставляем нейтральной.
"""
if not ts or not window_hours:
return None
age_hours = (time.time() - ts) / 3600.0
return round(age_hours / window_hours, 3)
def _funpay_link(item_name: str, game: str, stored_url: Optional[str]) -> str:
"""Ссылка на FunPay для Избранного (п.5 задачи): если оффер найден в
последнем скане — прямая ссылка на лот (stored_url из purchases); если нет
(например, товар давно не продавался на FunPay) — ссылка на поиск по
названию на странице категории этой игры, чтобы всё равно было куда
кликнуть."""
if stored_url:
return stored_url
base = funpay_urls.get(game, 'https://funpay.com/lots/1906/')
return f'{base}?query={quote_plus(item_name)}'
def build_favorites_rows() -> List[dict]:
rows = []
for p in purchases_list():
goods_id = p['goods_id']
current_price = None
price_source = '-'
price_updated = None
funpay_min_price_rub = None
live_row = ROW_REGISTRY.get(goods_id)
# "Окно анализа" самого товара — берём с живой строки сканера, если
# она есть в этой сессии; иначе — общий запасной порог (тот же, что
# используется как верхняя граница волатильности по умолчанию).
window_hours = (live_row or {}).get('history_window_hours') or VOLATILITY_HIGH_WINDOW_HOURS
if live_row and live_row.get('sell_price'):
current_price = live_row['sell_price']
price_source = live_row.get('source_label', '-')
price_updated = time.time()
else:
scan = cache_get_scan_result(goods_id)
if scan is not None and scan['price_rub']:
current_price = scan['price_rub']
price_source = SOURCE_LABELS.get(scan['source'], scan['source'])
price_updated = scan['updated_at']
else:
steam = steam_cache_get(goods_id)
if steam is not None and steam['price_rub']:
current_price = steam['price_rub']
price_source = 'Steam'
price_updated = steam['updated_at']
# Текущий минимум FunPay — для цветового сравнения со "своей" ценой
# продажи (п.3 задачи). Берём из живой строки сканера, если в ЭТОМ
# скане нашёлся оффер FunPay; иначе — из кэша scan_results, если
# последний известный источник для этого товара был именно FunPay.
if live_row and live_row.get('sell_source') == 'FunPay':
funpay_min_price_rub = live_row.get('sell_price')
else:
scan = cache_get_scan_result(goods_id)
if scan is not None and scan['source'] == 'funpay' and scan['price_rub']:
funpay_min_price_rub = scan['price_rub']
buff = buff_price_get(goods_id)
buy_price = p['buy_price_rub']
# "Моя цена продажи" (п.3 задачи) — редактируемая пользователем цена
# листинга. Для покупок, добавленных ДО этой версии (sell_price_rub
# ещё NULL в БД), используем текущую рыночную как разумный дефолт —
# иначе профит "Избранного" резко обнулился бы для всех старых записей.
sell_price_rub = p['sell_price_rub'] if p['sell_price_rub'] else current_price
profit_rub = profit_pct = None
if sell_price_rub:
_, payout = calc_payout_chain(sell_price_rub)
profit_rub = round(payout - buy_price, 2)
profit_pct = round((profit_rub / buy_price) * 100, 2) if buy_price > 0 else 0.0
buff_url = p['buff_url'] or f'https://buff.163.com/goods/{goods_id}?game={BUFF_GAME_CODES.get(p["game"], "csgo")}'
funpay_url = _funpay_link(p['item_name'], p['game'], p['funpay_url'])
rows.append({
'id': p['id'],
'goods_id': goods_id,
'purchased_at_str': _fmt_ts(p['purchased_at']),
'item_name': p['item_name'],
'item_name_html': f'<a href="{buff_url}" target="_blank" rel="noopener noreferrer">{p["item_name"]}</a>',
'game': p['game'],
'buy_price_rub': round(buy_price, 2),
'sell_price_rub': round(sell_price_rub, 2) if sell_price_rub else None,
'current_price_rub': round(current_price, 2) if current_price else None,
'funpay_min_price_rub': round(funpay_min_price_rub, 2) if funpay_min_price_rub else None,
'price_source': price_source,
'buff_now_rub': buff['price_rub'] if buff is not None else None,
'profit_rub': profit_rub,
'profit_pct': profit_pct,
'price_updated_str': _fmt_ts(price_updated),
'price_age_ratio': _age_ratio(price_updated, window_hours),
'buff_updated_str': _fmt_ts(p['buff_updated_at']),
'buff_age_ratio': _age_ratio(p['buff_updated_at'], window_hours),
'funpay_updated_str': _fmt_ts(p['funpay_updated_at']),
'funpay_age_ratio': _age_ratio(p['funpay_updated_at'], window_hours),
'funpay_html': f'<a href="{funpay_url}" target="_blank" rel="noopener noreferrer">FunPay ↗</a>',
'steam_updated_str': _fmt_ts(p['steam_updated_at']),
'steam_age_ratio': _age_ratio(p['steam_updated_at'], window_hours),
'delete_action': 'Удалить',
})
return rows
def refresh_favorites() -> List[dict]:
rows = build_favorites_rows()
# Не вызываем favorites_grid.update(): этот вызов заново обновляет весь
# NiceGUI-компонент AG Grid. В скрытой вкладке это могло происходить
# одновременно с перерасчётом размеров и приводило к визуальному
# "исчезновению" строк/колонок. Меняем rowData напрямую через AG Grid API.
favorites_grid.options['rowData'] = rows # сохраняем состояние на backend
try:
favorites_grid.run_grid_method('setGridOption', 'rowData', rows)
favorites_grid.run_grid_method('refreshCells', {'force': True})
except Exception:
logger.debug('Не удалось обновить favorites grid через AG Grid API', exc_info=True)
total_profit = sum(r['profit_rub'] for r in rows if r['profit_rub'] is not None)
favorites_summary_label.set_text(
f'Покупок: {len(rows)} | суммарный профит по текущим ценам: {total_profit:.2f} Р'
)
return rows
def on_favorites_cell_clicked(e) -> None:
args = e.args or {}
if args.get('colId') != 'delete_action':
return
data = args.get('data') or {}
if not data.get('id'):
return
_delete_target.clear()
_delete_target.update(data)
delete_dialog_label.set_text(f'Удалить «{data.get("item_name")}» из покупок?')
delete_dialog.open()
def on_favorites_cell_value_changed(e) -> None:
"""
Инлайн-редактирование «Цена покупки»/«Моя цена продажи» прямо в ячейке
(п.3 задачи) — сохраняет в SQLite и тут же пересчитывает профит через
полный refresh_favorites() (дешёво: одна выборка из purchases + расчёт
в Python, без сетевых запросов), а не точечный патч одной ячейки — так
гарантированно не разъезжаются профит/цвет подсветки с новым значением.
"""
args = e.args or {}
col_id = args.get('colId')
data = args.get('data') or {}
purchase_id = data.get('id')
if not purchase_id or col_id not in ('buy_price_rub', 'sell_price_rub'):
return
try:
new_value = float(args.get('newValue'))
except (TypeError, ValueError):
ui.notify('Введите число', type='warning')
refresh_favorites()
return
if new_value <= 0:
ui.notify('Цена должна быть больше 0', type='warning')
refresh_favorites()
return
if col_id == 'buy_price_rub':
purchase_update_prices(purchase_id, buy_price_rub=new_value)
else:
purchase_update_prices(purchase_id, sell_price_rub=new_value)
refresh_favorites()
def confirm_delete() -> None:
if _delete_target.get('id'):
purchase_delete(_delete_target['id'])
# Полный ресинк вместо точечного discard: у одного goods_id может быть
# несколько отдельных покупок (докупили тот же предмет ещё раз) — после
# удаления ОДНОЙ из них товар может всё ещё оставаться отслеживаемым.
_TRACKED_PURCHASE_IDS.clear()
_TRACKED_PURCHASE_IDS.update(purchased_goods_ids())
ui.notify(f'Удалено: {_delete_target.get("item_name")}', type='info')
delete_dialog.close()
refresh_favorites()
delete_confirm_btn.on_click(confirm_delete)
favorites_grid.on('cellClicked', on_favorites_cell_clicked)
favorites_grid.on('cellValueChanged', on_favorites_cell_value_changed)
favorites_refresh_btn.on_click(refresh_favorites)
_favorites_sync_generation = 0
async def _sync_favorites_after_tab_open(generation: int) -> None:
# Дожидаемся, пока Quasar реально покажет tab-panel и AG Grid получит
# ненулевой размер DOM. Это устраняет первый запуск, когда вкладка
# открывается уже с покупками, но grid был создан скрытым.
await asyncio.sleep(0.15)
if generation != _favorites_sync_generation:
return
refresh_favorites()
await asyncio.sleep(0.10)
if generation != _favorites_sync_generation:
return
try:
favorites_grid.run_grid_method('autoSizeAllColumns')
except Exception:
logger.debug('Не удалось пересчитать ширину favorites grid', exc_info=True)
def _on_tab_change(e) -> None:
global _favorites_sync_generation
value = e.value
value_name = getattr(value, 'value', value)
if value_name != 'favorites':
return
_favorites_sync_generation += 1
generation = _favorites_sync_generation
background_tasks.create(
_sync_favorites_after_tab_open(generation),
name='favorites_tab_sync',
)
tabs.on_value_change(_on_tab_change)
async def refresh_data():
global cny_rub_rate, funpay_urls, current_game, _refresh_in_progress
if _refresh_in_progress:
ui.notify('Обновление уже выполняется, подождите...', type='warning')
return
_refresh_in_progress = True
start_time = time.perf_counter()
first_rows_logged = False
ensure_steam_worker()
# Актуализируем набор отслеживаемых goods_id ДО начала обхода каталога —
# иначе build_row() не узнает, что для только что добавленной (в прошлую
# сессию) покупки нужно синхронизировать ссылки (см. sync_purchase_market_links).
global _TRACKED_PURCHASE_IDS
_TRACKED_PURCHASE_IDS = purchased_goods_ids()
try:
ui.notify('Запрос данных FunPay & Buff163...', type='info')
cny_rub_rate = await fetch_cny_rub_rate()
rate_label.set_text(f'Курс CNY: {cny_rub_rate} Р')
if not funpay_urls:
funpay_urls = await discover_funpay_urls()
min_rub = min_price_input.value or 0.0
min_sales = int(min_sales_input.value or 0)
lots_source = lots_source_select.value or 'Buff'
min_steam_sales = int(min_steam_sales_input.value or 0)
current_scan_filters.update(
lots_source=lots_source, min_sales=min_sales, min_steam_sales=min_steam_sales
)
steam_queue.clear()
ROW_REGISTRY.clear()
ROW_CONTEXTS.clear()
def push_partial_results(rows_so_far: List[dict]) -> None:
nonlocal first_rows_logged
visible = _visible_rows(rows_so_far)
_enqueue_rows_for_steam(visible)
grid.options['rowData'] = visible
grid.update()
grid.run_grid_method('autoSizeAllColumns')
if visible and not first_rows_logged:
first_rows_logged = True
logger.info('Первые товары в таблице через %.2f сек.', time.perf_counter() - start_time)
new_data = await fetch_market_data(
current_game, cny_rub_rate, min_rub, min_sales,
lots_source=lots_source,
min_steam_sales=min_steam_sales,
on_batch=push_partial_results,
)
push_partial_results(new_data)
elapsed = time.perf_counter() - start_time
ui.notify(
f'Загружено предметов: {len(_visible_rows(new_data))} | в Steam-очереди: {len(steam_queue)}',
type='positive',
)
logger.info('Скан завершён за %.2f сек.', elapsed)
refresh_favorites()
finally:
_refresh_in_progress = False
try:
auto_timer = ui.timer(
AUTO_REFRESH_INTERVAL_MINUTES * 60, refresh_data,
active=ENABLE_AUTO_REFRESH, immediate=False,
)
except TypeError:
auto_timer = ui.timer(
AUTO_REFRESH_INTERVAL_MINUTES * 60, refresh_data, active=ENABLE_AUTO_REFRESH
)
ui.timer(2.0, update_queue_status_label)
def on_interval_change(e):
try:
new_val = float(e.value)
if new_val >= 5:
auto_timer.interval = new_val * 60
except (ValueError, TypeError):
pass
def on_auto_refresh_toggle(e):
auto_timer.active = bool(e.value)
interval_input.on_value_change(on_interval_change)
auto_refresh_switch.on_value_change(on_auto_refresh_toggle)
async def on_game_change():
global current_game
current_game = game_select.value
await refresh_data()
game_select.on_value_change(on_game_change)
refresh_btn.on('click', refresh_data)
# Даём браузеру успеть смонтировать оба AG Grid до первого большого обновления.
# Сам favorites grid также синхронизируем отдельно чуть позже — это нужно для
# случая, когда покупки уже есть в SQLite до первого запуска программы.
ui.timer(0.5, refresh_data, once=True)
ui.timer(0.8, refresh_favorites, once=True)
ui.run(title='Arbitrage Scanner', port=8080, reload=False)