Skip to content

Instantly share code, notes, and snippets.

@rgardner

rgardner/alt_experiment_design.py

Created May 11, 2016 19:37
Show Gist options
  • Save rgardner/a353d289a715e03c4438d91831c62ee8 to your computer and use it in GitHub Desktop.
Save rgardner/a353d289a715e03c4438d91831c62ee8 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
alt_experiment_design.py
from datetime import datetime
import enum
from infoaggexperiment.database import db
import random
from sqlalchemy.ext.hybrid import hybrid_property
import uuid
NO_FEEDBACK_GROUP = 1
class Treatment(db.Model):
id = db.Column(db.Integer, primary_key=True)
profiles = db.relationship('Profile')
name = db.Column(db.String(10))
p = db.Column(db.Float, nullable=False)
q = db.Column(db.Float, nullable=False)
H = db.Column(db.Integer, nullable=False)
L = db.Column(db.Integer, nullable=False)
s = db.Column(db.Float, nullable=False)
def __repr__(self):
return '<Treatment {}>'.format(self.name)
class Profile(db.Model):
id = db.Column(db.Integer, primary_key=True)
treatment = db.Column(db.Integer, db.ForeignKey('treatment.id'))
reliability = db.Column(db.PickleType, nullable=False)
prob_unbias = db.Column(db.PickleType, nullable=False)
class Experiment(db.Model):
id = db.Column(db.Integer, primary_key=True)
session_id = db.Column(db.Integer, nullable=False)
treatments = db.relationship('Treatment', secondary=session_treatment_table)
profiles = db.relationship('Profile')
elections = db.relationship('Election')
group = db.Column(db.Integer, nullable=False)
subject_id = db.Column(db.Integer, db.ForeignKey('subject.id'))
subject = db.relationship('Subject', back_populates='experiment')
completed = db.Column(db.Boolean, defualt=False)
def __repr__(self):
return '<Session={}Group={}>'.format(self.session_id, self.group)
@property
def feedback(self):
"""Should the subject receive feedback."""
return self.group == NO_FEEDBACK_GROUP
@property
def current_election(self):
"""Return the latest not completed election."""
return Election.query.filter_by(session_id=self.session_id,
completed=False).first()
@property
def current_treatment(self):
"""Return the treatment in the current election."""
if self.current_election is not None:
return Treatment.query.get(self.current_election.treatment_id)
@hybrid_property
def last_vote_time(self):
return self.current_election.vote_time
class Vote(enum.Enum):
incorrect = 0
correct = 1
abstain = 2
class Election(db.Model):
id = db.Column(db.Integer, primary_key=True)
experiment_id = db.Column(db.Integer, db.ForeignKey('experiment.id'), nullable=False)
experiment = db.relationship('Experiment')
treatment_id = db.Column(db.Integer, db.ForeignKey('treatment.id'), nullable=False)
election_num = db.Column(db.Integer, nullable=False)
reliability = db.Column(db.Float, nullable=False)
state = db.Column(db.String(10), nullable=False)
info = db.Column(db.Integer, nullable=False)
is_biased = db.Column(db.Boolean)
vote = db.Column(db.Integer)
vote_time = db.Column(db.DateTime)
_group_decision = db.Column(db.Integer)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.state = random.choice(['red', 'blue'])
self.info = Vote.correct if random.random() < self.reliability else Vote.incorrect
# nonexperts are not biased. Experts are biased with probability 1 - s
treatment = Treatment.query.get(self.treatment_id)
if self.reliability == treatment.p:
# subject is an expert
self.is_biased = random.random() < 1 - treatment.s
else:
# subject is not an expert and cannot be biased
self.is_biased = False
def vote(self, choice):
# record vote
assert vote in ('red', 'blue', 'abstain')
if vote == self.correct_jar:
self.subject_player.votes.append(Vote.correct)
elif vote == self.incorrect_jar:
self.subject_player.votes.append(Vote.incorrect)
else:
self.subject_player.votes.append(Vote.abstain)
self.vote_time = datetime.utcnow()
check_completed(self.experiment)
def voters(self):
"""Collate the other voters in this election.
Get the current election in the other session-treatment-group.
If the group is the NO_FEEDBACK_GROUP, then we use the current group.
Otherwise, we use the previous group.
"""
Voter = namedtuple('Voter', 'reliability vote')
if self.experiment.group == NO_FEEDBACK_GROUP:
group = self.subject.experiment.group
else:
group = self.subject.experiment.group - 1
# experiments in same session and group
elections = Election.query.filter_by(session_id=self.session_id,
treatment_id=self.treatment_id,
group=group,
election_num=self.election_num).all()
return [Voter(e.reliability, e.vote) for e in elections]
@hybrid_property
def session_id(self):
return self.experiment.session_id
@hybrid_property
def group(self):
return self.experiment.group
@hybrid_property
def completed(self):
return self.vote is not None
@property
def group_decision(self):
if self._group_decision:
return self._group_decision
voters = self.voters()
if all(v.vote is not None for v in voters):
# only do it if everyone completed their votes
votes = [v.vote for v in voters]
self._group_decision = count_votes(votes).value
db.session.commit()
return self._group_decision
@property
def score(self):
if self.group_decision is None:
return 0
# If the subject is biased, then they are paid if their bias was
# chosen; unbiased subjects are paid if the group decision is correct.
if self.is_biased:
bias_correct = self.info == Vote.correct
group_correct = self.group_decision == Vote.correct
return 30 if bias_correct == group_correct else -70
return 30 if self.group_decision == Vote.correct else -70
class Subject(db.Model):
id = db.Column(db.Integer, primary_key=True)
hit_id = db.Column(db.String(120), nullable=False, unique=True)
worker_id = db.Column(db.String(120), nullable=False, unique=True)
assignment_id = db.Column(db.String(120), nullable=False, unique=True)
experiment_id = db.Column(db.Integer, db.ForeignKey('experiment.id'), unique=True)
experiment = db.relationship('Experiment')
survey_code = db.Column(
db.String(36),
unique=True,
default=lambda: str(uuid.uuid4()))
paid = db.Column(db.Boolean, default=False)
rewarded = db.Column(db.Boolean, default=False)
# Election functions
def count_votes(votes):
"""Return Vote.correct or Vote.incorrect.
This function is not pure; if the counts are equal, then
we choose correct/incorrect randomly.
"""
correct = votes.count(Vote.correct)
incorrect = votes.count(Vote.incorrect)
if correct > incorrect:
return Vote.correct
elif correct < incorrect:
return Vote.incorrect
else:
# choose correct/incorrect randomly
return Vote.correct if random.random() < 0.5 else Vote.incorrect
# Subject functions
def create_subject(hit_id, worker_id, assignment_id):
"""Create subject and associate with experiment entry.
postconditions: See assert_create_postconditions.
Raises
- ExperimentDone: all experiment entries completed
- ExperimentBusy: no available experiment.
"""
experiment = find_available_experiment()
# create new subject and associate with experiment entry
subject = Subject(hit_id=hit_id,
worker_id=worker_id,
assignment_id=assignment_id,
experiment=experiment)
experiment.subject = subject
assert_create_postconditions()
db.session.add(subject)
db.session.commit()
return subject
def find_available_experiment():
"""Find first available experiment.
Find first available experiment. An experiment is available if
all experiments in the previous group are completed.
preconditions: at least one available experiment
Raises:
- ExperimentDone: all experiment entries completed
- ExperimentBusy: no experiment available
"""
if all(e.completed for e in Experiment.query.all()):
raise ExperimentDone
availableq = Experiment.query.filter(Experiment.subject_id.isnot(None))
experiments = Experiment.query.filter(Experiment.subject_id.is_(None)).all()
for experiment in experiments:
if experiment.group == NO_FEEDBACK_GROUP:
return experiment
completed = availablq.filter_by(session_id=experiment.session_id,
group=experiment.group - 1).all()
if all(e.completed for e in completed.all()):
return experiment
if all(e.completed for e in Experiment.query.all()):
raise ExperimentDone
raise ExperimentBusy
def assert_create_postconditions(subject):
"""Assert create consistency.
- experiment is not available (has subject)
- experiment is not completed
- all elections have no vote
- all elections have no vote_time
- all elections have no _group_decision
- subject has an experiment assigned to him/her
"""
assert subject.experiment.subject_id is not None
assert subject.experiment.subject is not None
assert not subject.experiment.completed
elections = subject.experiment.elections
assert all(e.vote is None for e in elections)
assert all(e.vote_time is None for e in elections)
assert all(e._group_decision is None for e in elections)
assert subject.experiment_id is not None
assert subject.experiment is not None
def drop_subject(subject):
"""Drop subject from experiment.
postconditions: See assert_drop_postconditions
"""
assert subject.experiment_id is not None
# save these identifiers to assert
subject_id = subject.id
experiment_id = subject.experiment.id
# reset elections
for election in subject.experiment.elections:
election.vote = None
election.vote_time = None
election._group_decision = None
# reset experiment entry
subject.experiment.completed = False
subject.experiment.subject = None
# reset subject
subject.experiment = None
db.session.commit()
# assert postconditions
subject = Subject.query.get(subject_id)
experiment = Experiment.query.get(experiment_id)
assert_drop_postconditions(subject, experiment)
def assert_drop_postconditions(subject, experiment):
"""Assert drop consistency.
- experiment is available (no subject)
- experiment is not completed
- all elections have no vote
- all elections have no vote_time
- all elections have no _group_decision
- subject does not have an experiment assigned to him/her
"""
assert experiment.subject_id is None
assert experiment.subject is None
assert not experiment.completed
elections = experiment.elections
assert all(e.vote is None for e in elections)
assert all(e.vote_time is None for e in elections)
assert all(e._group_decision is None for e in elections)
assert subject.experiment_id is None
assert subject.experiment is None
def check_subject_inactive(subject):
if subject.experiment is not None:
time_since_last_vote = datetime.utcnow() - subject.Experiment.last_vote_time
if time_since_last_vote > app.timeout_threshold:
# subject is inactive and should be dropped
drop_subject(subject)
def mark_subject_completed(subject):
subject.experiment.completed = True
db.session.commit()
def check_group_status(session_id, group_id):
"""check whether all experiments with session,group are completed"""
experiments = Experiment.query.filter_by(session_id=session_id, group_id=group_id).all()
if all(experiment.completed for experiment in experiments):
calculate_results()
# Seed the database
# create 8 treatments
# create 120 experiments (session, treatments, profiles, groups)
# for each experiment
# create elections_per_treatment elections per treatment
# (election_num, bias, reliability)
# Experiment Flow
try:
subject = create_subject(hit_id=1, worker_id=1, assignment_id=1)
except SessionBusy:
print('Please come back in 45 minutes')
except ExperimentDone:
print('The experiment is now completed. Thank you for your interest')
while not subject.completed:
period = subject.create_period()
period.vote('red')
period.feedback()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment