Python Weekly #13
Cross Chain Python DCA Trading Bots & more
Hello world! Welcome to another edition of the Web3 Python Weekly. Each week we’ll bring you one exercise, tweet, & video to keep you up to date on all things happening at the crossroads of Web3 & python. Thanks for joining us!
Automating your DCA Strategy with Zhibai Zhang
What is a DCA Bot?
Dollar-cost-averaging (DCA) is a type of trading execution algorithms. It is commonly used in TradFi and has also gained popularity in crypto trading lately. The algorithm evenly distributes the overall investment dollar amount into smaller trades, and executes them separately at a predefined schedule.
For example, Alice has $10000 that she would like to invest in a stock. Instead of buying $10000 worth of shares, she uses a DCA algo and buys $1000 worth of shares once a day, over the course of 10 days.
Often DCA algos are implemented as a trading bot. A DCA bot is an automated trading tool that executes users’ trades following DCA algos programmatically.Why use DCA bots?
By executing through time, DCA algos can reduce investments’ volatility and lower potential market impact. When the size of the investment is large, using DCA becomes very beneficial as smaller trades perturb the market much less. Given the scheduling nature of the algorithm, deploying a bot to execute DCA trades is ideal. Volume’s DCA bot product:
Volume builds open source DCA bots on Paloma Chain. Fresh out of the oven, here's an example of setting up a DCA bot on Paloma Chain for PancakeSwap:# @version 0.3.7
struct Deposit:
depositor: address
path: DynArray[address, 8]
input_amount: uint256
number_trades: uint256
interval: uint256
starting_time: uint256
interface PancakeswapRouter:
def swapExactTokensForTokens(amountIn: uint256, amountOutMin: uint256, path: DynArray[address, 8], to: address, deadline: uint256): nonpayable
def getAmountsOut(amountIn: uint256, path: DynArray[address, 8]) -> DynArray[uint256, 7]: view
compass_evm: public(address)
admin: public(address)
deposit_list: HashMap[uint256, Deposit]
next_deposit: public(uint256)
PANCAKESWAP_ROUTER: constant(address) = 0x10ED43C718714eb63d5aA57B78B54704E256024E
@external
def __init__(_compass_evm: address):
self.compass_evm = _compass_evm
self.admin = msg.sender
@internal
def _safe_transfer_from(_token: address, _from: address, _to: address, _value: uint256):
_response: Bytes[32] = raw_call(
_token,
_abi_encode(_from, _to, _value, method_id=method_id("transferFrom(address,address,uint256)")),
max_outsize=32
) # dev: failed transferFrom
if len(_response) > 0:
assert convert(_response, bool), "failed transferFrom" # dev: failed transferFrom
@external
def deposit(path: DynArray[address, 8], input_amount: uint256, number_trades: uint256, interval: uint256, starting_time: uint256):
self._safe_transfer_from(path[0], msg.sender, self, input_amount)
_next_deposit: uint256 = self.next_deposit
self.deposit_list[_next_deposit] = Deposit({
depositor: msg.sender,
path: path,
input_amount: input_amount,
number_trades: number_trades,
interval: interval,
starting_time: starting_time
})
_next_deposit += 1
self.next_deposit = _next_deposit
@internal
def _safe_approve(_token: address, _to: address, _value: uint256):
_response: Bytes[32] = raw_call(
_token,
_abi_encode(_to, _value, method_id=method_id("approve(address,uint256)")),
max_outsize=32
) # dev: failed approve
if len(_response) > 0:
assert convert(_response, bool), "failed approve" # dev: failed approve
@external
def swap(swap_id: uint256, amount_out_min: uint256):
assert msg.sender == self.compass_evm, "not compass"
_next_deposit: uint256 = self.next_deposit
assert swap_id < _next_deposit
_deposit: Deposit = self.deposit_list[swap_id]
assert _deposit.number_trades > 0, "all traded"
assert _deposit.starting_time + _deposit.interval <= block.timestamp
_amount: uint256 = _deposit.input_amount / _deposit.number_trades
_deposit.input_amount -= _amount
_deposit.number_trades -= 1
_deposit.starting_time = block.timestamp
self._safe_approve(_deposit.path[0], PANCAKESWAP_ROUTER, _amount)
PancakeswapRouter(PANCAKESWAP_ROUTER).swapExactTokensForTokens(_amount, amount_out_min, _deposit.path, _deposit.depositor, block.timestamp)
if _deposit.number_trades == 0:
if swap_id < _next_deposit - 1:
self.deposit_list[swap_id] = self.deposit_list[_next_deposit - 1]
self.next_deposit = _next_deposit - 1
else:
self.deposit_list[swap_id] = _deposit
@view
@external
def triggerable_deposit() -> (uint256, uint256, uint256):
assert msg.sender == ZERO_ADDRESS
_size: uint256 = self.next_deposit
for i in range(1000000):
if i == _size:
return 0, 0, 0
_deposit: Deposit = self.deposit_list[i]
if _deposit.starting_time + _deposit.interval <= block.timestamp:
_amount: uint256 = _deposit.input_amount / _deposit.number_trades
_out_amount: DynArray[uint256, 7] = PancakeswapRouter(PANCAKESWAP_ROUTER).getAmountsOut(_amount, _deposit.path)
return i, _out_amount[len(_out_amount) - 1], _deposit.number_trades
return 0, 0, 0How can developers get started?
Visit the Volume website to know more about what we're building & don't hesitate to reach out if you have something in mind.
Website
Email
Github
If you want to go waaay into it:
And make use of the latest bot to build bots:
That's it for today & see ya next time! If you liked this issue, be sure to share! 🐍
Gm Gm, Really loving this piece and the work you do here. I also run a web3 news substack for underrepresented creators called Facesofweb3. Would you be open to a recommendation exchange? Our subscribers need to be able to find each other!
Great job, once again!