Source code for swab

# encoding=UTF-8

# Simple WSGI A/B testing.
# (c) 2010-2018 Oliver Cope.
# See ``README.rst`` for usage instructions etc.

import base64
import logging
import math
import os
import re
from collections import OrderedDict
from random import Random
from struct import unpack_from
from time import time
from hashlib import md5
from functools import partial
    from urllib.parse import quote_plus
except ImportError:
    from urllib import quote_plus
from pkg_resources import resource_filename
from fresco import FrescoApp, Request, Response
from fresco.cookie import Cookie
from fresco_static import StaticFiles
from fresco_template import Piglet
from piglet import TemplateLoader

logger = logging.getLogger(__name__)
templateloader = TemplateLoader([resource_filename(__name__, 'templates')])
render = Piglet(templateloader)
ustr = type(u'')

__version__ = '0.2.2'

is_bot_ua = re.compile("|".join([
    r".*indy library",
    r".*jakarta Commons",
]), re.I).match

[docs]class Swab(object): """ Simple WSGI A/B testing """ def __init__(self, datadir, wsgi_mountpoint='/swab'): """ Create a new Swab test object :param datadir: Path to data storage directory :param wsgi_mountpoint: The path swab-specific views and resources will be served from by the middleware. """ self.datadir = datadir #: Mapping of {<experiment name>: <Experiment object>} self.experiments = {} #: Mapping of {<goal name>: [<Experiment object>, ...]} self.experiments_by_goal = {} self.wsgi_mountpoint = wsgi_mountpoint makedir(self.datadir) def include(self, environ, experiment): e = self.experiments[experiment] return e.include(environ) and not e.exclude(environ)
[docs] def middleware(self, app, cookie_domain=None, cookie_path=None, cache_control=True): """ Middleware that sets a random identity cookie for tracking users. The identity can be overwritten by setting ``environ['']`` before start_response is called. On egress this middleware will then reset the cookie if required. :param app: The WSGI application to wrap :param cookie_domain: The domain to use when setting cookies. If ``None`` this will not be set and the browser will default to the domain used for the request. :param cache_control: If ``True``, replace the upstream application's cache control headers for any request where show_variant is invoked. """ swabapp = FrescoApp() static_files = StaticFiles() static_files.init_app(swabapp) static_files.add_package(__name__, 'static') swabapp.route_wsgi('/results', self.results_app) swabapp.route_wsgi('/r.js', self.record_trial_app) def middleware(environ, start_response): environ['swab.swab'] = self environ[''] = initswabid = _getswabid(environ) environ['swab.invoked'] = False environ['swab.experiments_invoked'] = set() if initswabid is None: environ[''] = generate_id() if environ['PATH_INFO'][:len(self.wsgi_mountpoint)] == \ self.wsgi_mountpoint: environ['SCRIPT_NAME'] += self.wsgi_mountpoint environ['PATH_INFO'] = \ environ['PATH_INFO'][len(self.wsgi_mountpoint):] return swabapp(environ, start_response) def my_start_response(status, headers, exc_info=None, _cache_headers=set(['cache-control', 'expires', 'etag', 'last-modified'])): if not environ['swab.invoked']: return start_response(status, headers, exc_info) swabid = _getswabid(environ) if swabid == initswabid and swabid is not None: return start_response(status, headers, exc_info) if swabid is None: swabid = generate_id() environ[''] = swabid _cookie_path = (cookie_path or environ.get('SCRIPT_NAME') or '/') cookie = Cookie('swab', swabid.decode('ascii'), path=_cookie_path, domain=cookie_domain, httponly=True, max_age=365) if cache_control and environ.get('swab.experiments_invoked'): headers = [(k, v) for k, v in headers if k.lower() not in _cache_headers] headers.append(('Cache-Control', 'no-cache')) headers.append(("Set-Cookie", str(cookie))) return start_response(status, headers, exc_info) return app(environ, my_start_response)
return middleware def add_goals(self, names): for n in names: self.experiments_by_goal.setdefault(n, set()) def add_experiment(self, name, variants=None, goal=None): exp = self.experiments[name] = Experiment(name) if variants: exp.add(*variants) goal = goal if goal is not None else name self.add_goals({goal}) self.experiments_by_goal[goal].add(exp) makedir(os.path.join(self.datadir, name)) return exp def results_app(self, environ, start_response): request = Request(environ) dedupe = not bool(request.get('nodedupe')) data = self.collect_experiment_data(dedupe=dedupe) for exp in data: vdata = data[exp]['variants'] control = request.get('control.' + exp, self.experiments[exp].control) control_trials = vdata[control]['trials'] control_goals = vdata[control]['goals'] data[exp]['control'] = control for variant in vdata: variant_trials = vdata[variant]['trials'] for goal, goaldata in vdata[variant]['goals'].items(): control_rate = control_goals[goal]['rate'] control_conversions = control_goals[goal]['conversions'] z = zscore(goaldata['rate'], variant_trials, control_rate, control_trials) goaldata['z'] = z goaldata['confidence'] = cumulative_normal_distribution(z) goaldata['p_beats_control'] = probability_b_beats_a( control_trials, control_conversions, variant_trials, goaldata['conversions'] ) return render.as_response( resource_filename('swab', 'templates/results.html'), { 'request': request, 'experiments': self.experiments.values(), 'data': data, 'dedupe': dedupe, } )(environ, start_response) def record_trial_app(self, environ, start_response): request = Request(environ) experiment = request.query.get('e') try: record_trial(environ, experiment) except KeyError: pass return Response([], cache_control="no-cache")(environ, start_response)
[docs] def collect_experiment_data(self, dedupe=False): """ Collect experiment data from the log files Return a dictionary of:: {<experiment>: { 'goals': [goal1, goal2, ...], 'variants': { 'v1': { 'trials': 1062, 'goals': { 'goal1': {'conversions': 43, 'rate': 0.0405}, 'goal2': {'conversions': 29, 'rate': 0.0273}, } }, ... } } """ data = {} for exp in self.experiments.values(): expdir = os.path.join(self.datadir, goals = sorted( [goal for goal, experiments in self.experiments_by_goal.items() if exp in experiments]) data[] = {'goals': goals, 'variants': {}} for variant in exp.variants: path = partial(os.path.join, expdir, variant) if dedupe: trial_identities = get_identities(path('__all__')) trialc = len(trial_identities) else: trialc = count_entries(path('__all__'), dedupe=False) data[]['variants'][variant] = vdata = {'trials': trialc, 'goals': {}} for goal in goals: vdata['goals'][goal] = goaldata = {} if dedupe: conv_identities = trial_identities.intersection( get_identities(path(goal))) convc = len(conv_identities) else: convc = count_entries(path(goal), dedupe=False) goaldata['conversions'] = convc goaldata['rate'] = (float(convc) / trialc if trialc else float('nan'))
return data class Experiment(object): invalid_chars = r'\\/:,' def __init__(self, name, seed_strategy=None): if not self._is_valid_name(name): raise ValueError("Experiment name may not contain any of {!r}" .format(set(self.invalid_chars))) = name #: The control variant name self.control = None #: Mapping of variant name → weight self.variants = OrderedDict() #: The default variant (typically shown to bots) self.default_variant = None #: Function returning a the RNG seed to use self.seed_strategy = (seed_strategy if seed_strategy is not None else default_seed_strategy) def add(self, *variants): """ Add variants. Each argument may be either a variant name, or a tuple of (variant-name, weight). For example: colors = Experiment('colors') colors.add('blue', 'red') # equal split between blue/red colors.add(('blue', 1), ('red', 3)) # red 3x more likely than blue """ for v in variants: if isinstance(v, (str, ustr)): name, weight = v, 1 else: name, weight = v if not self._is_valid_name(name): raise ValueError("Variant name may not contain any of {!r}" .format(set(self.invalid_chars))) if self.control is None: self.control = name self.variants[name] = self.variants.get(name, 0) + 1 self._recalculate() def _recalculate(self): """ Recalculate weighted_variants and default_variant """ self.weighted_variants = [n for n in self.variants for i in range(self.variants[n])] default = self.control default_weight = self.variants[self.control] for name, weight in self.variants.items(): if weight > default_weight: default, default_weight = name, weight self.default_variant = default def _is_valid_name(self, name): return not any(c in name for c in self.invalid_chars) def include(self, environ): return True def exclude(self, environ): return False def _getswabid(environ): """ Return the unique identifier from the WSGI environment if present, otherwise return ``None``. """ try: return environ[''] except KeyError: pass cookie = Request(environ).cookies.get('swab') if cookie: swabid = cookie.value.encode('ascii') environ[''] = swabid return swabid return None
[docs]def generate_id(urandom=os.urandom, encode=base64.b64encode): """ Return a unique id """
return encode(urandom(12)).strip()
[docs]def get_rng(environ, experiment, swabid): """ Return a random number generator with a fixed seed based on the current session's swabid """ r = Random() r.seed(experiment.seed_strategy(environ, experiment, swabid))
return r
[docs]def get_seed_from_bytes(s): """ Given a byte string, return a RNG seed value """
return unpack_from('l', md5(s).digest())[0] def default_seed_strategy(environ, experiment, swabid): return get_seed_from_bytes(swabid +'UTF-8'))
[docs]def show_variant(environ, experiment, record=False, variant=None): """ Return the variant name that ``environ`` is assigned to within ``experiment`` If ``record`` is true, write a line to the log file indicating that the variant was shown. (No deduping is done - the log line is always written. A page with ``show_variant`` might record multiple hits on reloads etc) :param experiment: Name of the experiment :param environ: WSGI environ :param record: If ``True``, record a trial for the experiment in the log file :param variant: force the named variant. Use this if your application wants to choose the variant based on some other criteria (eg SEO a/b testing where you assign the variant based on the URL) """ swab = environ['swab.swab'] exp = swab.experiments[experiment] variants = exp.weighted_variants swabid = _getswabid(environ) if not swab.include(environ, experiment): variant = variant if variant is not None else exp.control # Make sure bots don't receive randomized variants. Google's advice # (as of August 2017): show the version the majority of users see. # Failing that, show a consistent version (don't randomize) # if is_bot(environ): variant = variant if variant is not None else exp.control if variant is None: request = Request(environ) variant = request.query.get('swab.' + experiment) if variant is not None and variant in variants: return variant # Get a random int in the range 0 ≤ x < len(variants) # # We do this in preference to calling random.choice because # it guarantees a particular property of the sampling if the list of # variants changes. # # For example, given a list of variants ``['a', 'b']`` # we will choose a variant according to the output of # `r = rng.random()` such that # # 0.0 ≤ r < 0.5 → 'a' # 0.5 ≤ r < 1.0 → 'b' # # if later we find that 'a' is winning and want to exploit that # variant, we could change the list of variants to # ``['a', 'a', 'a', 'b']``, and the mapping would become: # # 0.0 ≤ r < 0.75 → 'a' # 0.75 ≤ r < 1.0 → 'b' # # Notice that the range corresponding to variant 'a' completely # contains the old values - ie users who previously saw the winning # variant 'a' will continue to see that variant. r = int(get_rng(environ, exp, swabid).random() * len(variants)) variant = variants[r] if variant not in variants: raise ValueError("Invalid variant {!r}. Choices are: {!r}" .format(variant, variants)) environ['swab.invoked'] = True invoked = environ.setdefault('swab.experiments_invoked', set()) if experiment in invoked: return variant invoked.add(experiment) if not record or is_bot(environ): return variant path = os.path.join(swab.datadir, experiment, variant, '__all__') try: f = open(path, 'a') except IOError: makedir(os.path.dirname(path)) f = open(path, 'a') try: f.write(_logline(swabid)) finally: f.close()
return variant record_trial = partial(show_variant, record=True) def record_trial_tag(environ, experiment=None): experiments = environ.get('swab.experiments_invoked', frozenset()) if experiment is None: if len(experiments) != 1: raise ValueError("record_trial_tag can't guess the experiment " "name, as show_variant has been called multiple " "times (or not at all). " "Fix this by passing an experiment name. ") experiment = next(iter(experiments)) if experiment not in experiments: logger.warn( "record_trial_tag called without a corresponding show_variant " "(experiment=%r)", experiment) request = Request(environ) if 'swab.' + experiment in request.query: return '' swab = environ['swab.swab'] return ('<script>' '(function(f,o,x){{' 'x=f.getElementsByTagName(o)[0],' 'o=f.createElement(o),' 'o.async=1,' 'o.src="{0}?e={1};s={2}";' 'x.parentNode.insertBefore(o,x)}})(document,"script")' '</script>').format( request.make_url(path=swab.wsgi_mountpoint + '/' + 'r.js', query=''), quote_plus(experiment), _getswabid(environ))
[docs]def is_bot(environ, is_bot_ua=is_bot_ua): """ Return True if the request is from a bot. Uses rather simplistic tests based on user agent and header signatures, but should still catch most well behaved bots. """ if is_bot_ua(environ.get('HTTP_USER_AGENT', '')): return True if 'HTTP_ACCEPT_LANGUAGE' not in environ: return True
return False def _logline(swabid): return '%-14.2f:%s\n' % (time(), swabid) def _logentries(path): with open(path, 'r') as f: for line in f: try: t, id = line.strip().split(':') except ValueError: continue yield float(t.strip()), id
[docs]def record_goal(environ, goal, experiment=None): """ Record a goal conversion by adding a record to the file at ``swab-path/<experiment>/<variant>/<goal>``. If experiment is not specified, all experiments linked to the named goal are looked up. This doesn't use any file locking, but we should be safe on any posix system as we are appending each time to the file. See for a discussion of the issue. """ if is_bot(environ): return swab = environ['swab.swab'] if experiment is None: try: experiments = swab.experiments_by_goal[goal] except KeyError: raise ValueError("Invalid goal: {!r}".format(goal)) else: experiments = [swab.experiments[experiment]] for experiment in experiments: if not swab.include(environ, continue variant = show_variant(environ,, record=False) path = os.path.join(swab.datadir,, variant, goal) try: f = open(path, 'a') except IOError: makedir(os.path.dirname(path)) f = open(path, 'a') try: f.write(_logline(_getswabid(environ))) finally:
[docs]def makedir(path): """ Create a directory at ``path``. Unlike ``os.makedirs`` don't raise an error if ``path`` already exists. """ try: os.makedirs(path) except OSError: # Path already exists or cannot be created if not os.path.isdir(path):
[docs]def count_entries(path, dedupe=True): """ Count the number of entries in ``path``. :param dedupe: if True, dedupe Entries so only one conversion is counted per identity. """ if dedupe: return len(get_identities(path)) else: if not os.path.exists(path): return 0 with open(path, 'rb') as f:
return sum(1 for line in f)
[docs]def get_identities(path): """ Return a Counter for identity entries in ``path`` """ if not os.path.isfile(path): return set()
return set(identity for t, identity in _logentries(path))
[docs]def zscore(p, n, pc, nc): """ Calculate the zscore of probability ``p`` over ``n`` tests, compared to control probability ``pc`` over ``nc`` tests See """ from math import sqrt try: return (p - pc) / sqrt( (p * (1 - p) / n) + (pc * (1 - pc) / nc)) except (ZeroDivisionError, ValueError):
return float('nan')
[docs]def cumulative_normal_distribution(z): """ Return the confidence level from calculating of the cumulative normal distribution for the given zscore. See and """ from math import exp b1 = +0.319381530 b2 = -0.356563782 b3 = +1.781477937 b4 = -1.821255978 b5 = +1.330274429 p = +0.2316419 c = +0.39894228 if z >= 0.0: t = 1.0 / (1.0 + p * z) return 1.0 - c * exp(-z * z / 2.0) * t * \ (t * (t * (t * (t * b5 + b4) + b3) + b2) + b1) else: t = 1.0 / (1.0 - p * z) return c * exp(-z * z / 2.0) * t * \
(t * (t * (t * (t * b5 + b4) + b3) + b2) + b1) def probability_b_beats_a(trials_a, goals_a, trials_b, goals_b): a_A = 1 + goals_a b_A = 1 + trials_a - goals_a a_B = 1 + goals_b b_B = 1 + trials_b - goals_b # Happens when goals > trials. # In this case the gamma function isn't defined and the remaining code # errors. if b_A <= 0 or b_B <= 0: return float('nan') total = 0.0 log = math.log exp = math.exp # Define a log beta function in terms of gamma lbeta = lambda a, b, lgamma=math.lgamma: (lgamma(a) + lgamma(b) - lgamma(a + b)) for i in range(a_B): total += exp( lbeta(a_A + i, b_B + b_A) - log(b_B + i) - lbeta(1 + i, b_B) - lbeta(a_A, b_A) ) return total