Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Addressing Issue #38 - Added Staging support #39

Merged
merged 1 commit into from
Aug 22, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
.env
.log
__pycache__
.pytest_cache
src/tempCodeRunnerFile.py
Expand Down
Empty file removed slippage_project.log
Empty file.
61 changes: 33 additions & 28 deletions src/price_providers/endpoint_auction_pricing.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,9 @@ class AuctionPriceProvider(AbstractPriceProvider):
"""Fetch auction prices."""

def __init__(self) -> None:
self.endpoint_url = {
"prod": f"https://api.cow.fi/mainnet/api/v1/solver_competition/by_tx_hash/"
self.endpoint_urls = {
"prod": f"https://api.cow.fi/mainnet/api/v1/solver_competition/by_tx_hash/",
"barn": f"https://barn.api.cow.fi/mainnet/api/v1/solver_competition/by_tx_hash/",
}

@property
Expand All @@ -19,32 +20,36 @@ def name(self) -> str:
def get_price(self, price_params: dict) -> float | None:
"""Function returns Auction price from endpoint for a token address."""
token_address, tx_hash = extract_params(price_params, is_block=False)
url = self.endpoint_url["prod"] + tx_hash
try:
response = requests.get(url)
response.raise_for_status()
data = response.json()

# Search for the token address in the auction prices
auction_prices = data.get("auction", {}).get("prices", {})
price = auction_prices.get(token_address.lower())

if price is None:
logger.warning(
f"Price for token {token_address} not found in auction data."
for environment, url in self.endpoint_urls.items():
try:
# append tx_hash to endpoint
response = requests.get(url + tx_hash)
response.raise_for_status()
data = response.json()

# Search for the token address in the auction prices
auction_prices = data.get("auction", {}).get("prices", {})
price = auction_prices.get(token_address.lower())

if price is None:
logger.warning(
f"Price for token {token_address} not found in auction data."
)
return None
# calculation for converting auction price from endpoint to ETH equivalent per token unit
price_in_eth = (float(price) / 10**18) * (
10 ** get_token_decimals(token_address) / 10**18
)
return None
# calculation for converting auction price from endpoint to ETH equivalent per token unit
price_in_eth = (float(price) / 10**18) * (
10 ** get_token_decimals(token_address) / 10**18
)
return price_in_eth

except requests.exceptions.RequestException as req_err:
logger.error(f"Error occurred during request: {req_err}")
except KeyError as key_err:
logger.error(f"Key error: {key_err}")
except Exception as e:
logger.error(f"An unexpected error occurred: {e}")
return price_in_eth

except requests.exceptions.HTTPError as err:
if err.response.status_code == 404:
pass
except requests.exceptions.RequestException as req_err:
logger.error(f"Error occurred during request: {req_err}")
except KeyError as key_err:
logger.error(f"Key error: {key_err}")
except Exception as e:
logger.error(f"An unexpected error occurred: {e}")

return None
Loading