Life is full of choices, but sometimes our intentions are out of sync with the opportunities to act on them.
“I’m gonna buy the dip!” I might say, sipping my coffee. But then I get a call from Mom. And while we’re talking, my favorite dog coin plunges 50% and remains there for several minutes until Elon tweets about it again, which restores the price.
If I had a system in place to automatically execute per my intention, I could have turned that dip into extra tokens. But alas, the meatverse trumped the metaverse and now that buying opportunity is just a candle on some chart.
Aligning Intent With Action
One of the projects I built was a simple Limit Order Bot that would perform swaps whenever certain price targets were observed at a particular pool.
It was fun, though limited and amateurish, but it demonstrated that you could program a bot that executed some action when a condition was met.
I saw a friend lamenting in Telegram that he couldn’t find a platform that allowed him to set a trailing stop loss on some memecoin. For those unfamiliar, a trailing stop loss is set below the current market price by some offset, either a fixed amount or a percentage. As the market price increases, so does the trailing stop. If the market price decreases to the level where the trailing stop is met, the position can be sold. Theoretically, this allows you to ride the price of an appreciating asset upward, then automatically reduce exposure if it depreciates by some amount.
It’s common that you can find a token on one platform, and a desirable feature on another, with no way to combine them. I got to thinking about it, remembered my old limit order bot, and became interested in the topic again.
What if we could build an execution engine of our own? By hooking a few pieces together, my friend could have his trailing stop loss on whatever memecoin he wanted, in whatever pool he wanted, on any chain.
Sounds like a winner!
The Plan
In order to implement some generic conditional action, we need to:
Identify required objects
Build action steps and conditions
Orchestrate the process
For now, I will limit the scope to a simple price-based trigger that sends a notification. Once the concept has been proven, more sophisticated conditions and actions can be developed.
Identify Required Objects
We care primarily about token price for this conditional action. At minimum, an ERC-20 token must be involved. It must have an associated price that can be monitored and updated.
The degenbot module provides an Erc20Token
class and a ChainlinkPriceContract
class that will be useful here.
Build Action Steps and Conditions
The notification will be a simple console notification, displayed by a call to print()
.
An interesting question: what should be responsible for that notification? We must be careful not to entangle things. An Erc20Token
object is responsible for retrieving and expressing the characteristics of an ERC-20 token contract. It determines the attributes you’d expect, like the token name, symbol, number of decimal places. It also exposes helper functions to retrieve the balance of an account, approvals set for various spenders on a particular account, and the total supply for the token.
It also stores a reference to an external price oracle. But other than expecting the oracle to provide an interface update_price()
that returns a float-formatted USD price, that oracle could be anything. It could be a Chainlink contract, or an external HTTP API like CoinGecko.
We don’t want Erc20Token
to be responsible for evaluating conditionals or executing on them, that’s far outside of its scope. Similarly, ChainlinkPriceContract
has no business doing that.
Orchestrate The Process
It’s clear that conditional action is a concept that should be encapsulated properly in a specialized class that provides the smarts when given the parts.
Let’s give the class a name: ConditionalAction
.
conditional_action.py
from typing import Any, Callable, Sequence
class ConditionalAction:
def __init__(
self,
condition: Callable[[Any], bool],
actions: Sequence[Callable[[Any], Any]],
):
self.condition = condition
self.actions = actions
The :Callable[]
stuff is type-hinting syntax for callables. A callable is an object that can be called with the ()
operator, which places it on the stack for execution by the interpreter. The format for the type hint is Callable[[argument types], return type]
.
My type hints above say that a ConditionalAction
object must be provided with a sequence of action
callables, and a condition
callable. I’ll define both soon, but the basic idea is that I have some condition
function that will return True
if the necessary requirements are met to begin execution of the action
callables. The return value of action
is not used and therefore irrelevent, so its return type is set to Any.
The ConditionalAction
object doesn’t care about the internal details of condition
or action
, only that both are callable, and that condition
returns a bool.
Now define a simple method to evaluate condition
and perform all of the action
if allowed:
def check(self):
if self.condition() is True:
for action in self.actions
action()
To try this out, run a simple interactive test on the Python console:
>>> from conditional_action import ConditionalAction
>>> def print_success():
print("success!")
...
>>> def _true():
return True
...
>>> action = ConditionalAction([print_success], _true)
>>> action.check()
success!
This demonstrates the approach cleanly: set up a conditional action, check the condition, if true then do the action. This just prints but you can imagine many actions that can be done very simply by a simple boolean check.
Dynamic Conditions
For standalone action
and condition
callables, the above is sufficient. But what if either callable is dynamic, depending on some fluctuating external state? Imagine the trailing stop loss example from earlier — the condition
callable would need to evaluate the current price for the token and its trailing stop. Getting the current price is no big deal, but condition
has no idea where the trailing stop should be, because it was not involved with setting it initially, or adjusting it as the price fluctuated.
So who is responsible for proper adjustment of the trailing stop?
We must consider our options carefully!
The responsibility of a ConditionalAction
is to check a condition
and then perform an action
. If state has to be maintained to properly evaluate the condition
, that state should be self-contained and otherwise maintained “by others”.
OK, so how do we implement self-contained state and provide it to the condition
?
I will use a technique here called dependency injection, whereby everything that the condition
callable needs is given to it directly during construction.
As an example, let’s build a callable with a very rough trailing stop loss check:
def price_below_trailing_stop(
token: Erc20Token,
state,
):
if token.price < state.trailing_stop:
return True
else:
return False
When this function is defined, it expects two inputs: token
and state
. I have defined the type for token
as an Erc20Token
, which implies that it has an associated price oracle that maintains it current price via the .price
attribute.
Further, it expects that the state
(which has no type hint) provides an attribute .trailing_stop
, which can be compared to the current price. The type is not particularly important, only the interface it exposes so the condition can be evaluated. This approach is commonly known as “duck typing” in Python — if it looks like a duck, and .quack()
s like a duck, it’s a duck.
As far as the token
and state
, those objects are maintained by others.
This callable can therefore be built at runtime and provided to the constructor of ConditionalAction
. As long as the token
and state
objects are properly tracked and updated to follow the live system, the condition
check will work as expected.
An Aside About Partial Functions
The functools module provides a function called partial()
that returns a callable object with certain arguments already pre-set. It’s useful when you want to provide a pre-loaded argument (such as an injected dependency) to a function without having to use complex wrappers.
If I wanted to include print("foo")
as an action
, I could not do that because actions=[print("foo")]
will call that print function immediately, and place its return value (None
) into the actions sequence. Definitely not what you want!
Instead, actions=[functools.partial(print, "foo")]
results in a callable object with that argument pre-loaded. This called is stored inside actions
, and will work as expected when executed later.
Price-Watcher Example
Getting back to the original goal, let’s build a simple watcher with a pair of ConditionalAction
s that print notifications when the Chainlink price of ETH (in USD) moves up or down.
We will use the Chainlink Ethereum mainnet oracle to observe the ETH/USD price feed contract at 0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419. Though this could be anything compatible with your particular condition
.