I have a loop that cycles days and discretised time slots of a financial time series, and some code for my strategy lies inside it. I extracted this inner code to a standalone function, to be able to use it in specific situations for testing, to only find out a bit later that this operation heavily increased the total computational time of this loop. It makes sense, the long function signature, the call, (stack memory?) all this has to introduce some overhead.
# Fast, but cannot use my strategy for specific instances:
for d in data:
# my strategy code
# Flexible but really slow
def my_strategy(**params):
# my strategy code
f = partial(my_strategy, **params)
for d in data:
f(d)
Now, my question: is there a way to keep both setups, the standalone function and the one with the loops and the actual code inside without explicit redundancy? I would need to have both.
I've been experiment with inspect.getsource(my_function) and wrap a copy of the outer function to then use eval but after 5 min I felt like an idiot, it does not feel right at all.
What's the best practice to go about this? If I duplicate the code, I have to make sure that if I modify one version, the other one is always in sync. I don't like this. I would like to mirror it efficiently.
As you might have thought, one option could be: leave the code where it is, inside the loop, and wrap your specific case with an array and feed it to the loop, whenever you need it. Well, I can't do that, mainly because I need to have a timed control over the inner code execution (am porting the back-tested strategy to real-time). Here I am just looking for some quick trick to mirror the code elegantly, if there is any.
--- EDIT ---
Ok so, to test this further, I implemented that horrible hack I mentioned and I made it work with exec. Thanks to this, I can now guarantee that the code used is exactly the same. The time difference observed is the same though. The loop that contains the code takes roughly HALF of the time taken by the loop with function call.
# horrible hack, please forgive me
def extract_strategy_function():
# Start
source_code = inspect.getsource(fastest_compute_trades)
# Cutting the code snippet
source_code = source_code.split('[safe_start - 1:]:', 1)[1].rsplit("if s['pos_open']:", 1)[0]
# Needs re-indentation so the parser doesn't complain
source_code = reindent(source_code, -2)
function_name = 'single_apply_strategy' # internal, no-one will see this anyway
function_string = f"""
def {function_name}(
# INDICATORS (non mutable)
i_time, close_price, r2, ma1, ma2, rsi, upper, lower,
# STATE VARIABLES, TX HISTORY (mutable, 2 dictionaries and 1 list of dicts)
s, gs, transactions,
# CONSTANT PARAMETERS (non mutable)
min_r2, macd_threshold, rsi_bound, waiting_slots, stop_loss,
take_profit, trail_profit, double_down, dd_qty, commission_fee,
# Day ID, useful when back-testing/grid-searching, left at the end so it can default to 0
i_day=0
):
{source_code}
""".strip()
# Evaluate it
exec(function_string) # Now in locals
return locals()[function_name]
def slow(...):
...
apply_strategy = partial(extract_strategy_function(),
**params, commission_fee=commission_fee, rsi_bound=rsi_bound)
for i_day, day in enumerate(data):
for i_time, close_price, r2, ma1, ma2, rsi, upper, lower, in list(enumerate(day))[safe_start - 1:]:
apply_strategy(i_time, close_price, r2, ma1, ma2, rsi, upper, lower,
day_state, global_state, transactions, i_day=i_day)
if day_state['pos_open']:
...
...
def fast(...,
#some extra parameters are now unpacked here,
#since all code is here they are now needed here
):
...
for i_day, day in enumerate(data):
for i_time, close_price, r2, ma1, ma2, rsi, upper, lower, in list(enumerate(day))[safe_start - 1:]:
# actual code contained in single_apply_strategy written here, the whole strategy (one timestep)
if day_state['pos_open']:
...
...
Any suggestions?
p.s. Needless to say, the output of the two setups is exactly the same
